From 7e484ba4e9ad4fb3f4c2b8d65d07ef427558c57e Mon Sep 17 00:00:00 2001 From: liuhaodong Date: Wed, 23 Sep 2026 10:23:31 +0800 Subject: [PATCH] =?UTF-8?q?fix=EF=BC=9A=E7=9B=B4=E6=92=AD=E4=BC=98?= =?UTF-8?q?=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 69 ++++++++++ cache/cache_key.go | 11 +- common/busi_error.go | 2 + common/mysql.go | 8 +- config-prod.yaml | 3 + handler/live_handler.go | 9 +- model/media.go | 1 + mqtt/message.go | 2 - mqtt/mqtt.go | 12 +- service/billing_service.go | 14 +- service/command_service.go | 83 +++++++++++- service/live_service.go | 269 +++++++++++++++++++++++++++---------- service/mqtt_subscriber.go | 44 ++++-- sql/001_schema.sql | 5 +- vo/media_vo.go | 18 ++- 15 files changed, 454 insertions(+), 96 deletions(-) diff --git a/README.md b/README.md index 9112f1e..c2a38f1 100644 --- a/README.md +++ b/README.md @@ -85,6 +85,75 @@ to a JSON-safe `providerPayload` data URI (`data:image/png;base64,...`); clients should use that value directly as the image `src` and must not UTF-8 decode the original response body. +## Live session device event uniqueness + +`live_session.uk_live_session_device_event` was originally built directly on +`device_event_id`. MySQL unique indexes treat multiple `NULL` values as +distinct but multiple empty strings as duplicates, so only one live session +without a reported device event could exist at a time. Creating a second live +session then failed with: + +``` +Error 1062 (23000): Duplicate entry '' for key 'live_session.uk_live_session_device_event' +``` + +Run `sql/023_live_session_device_event_nullable.sql` on any database created +before this fix. It adds a `device_event_key` generated column +(`NULLIF(device_event_id, '')`) and moves the unique index onto it. The +migration is repeatable. `sql/001_schema.sql` already contains the fixed +schema for new databases. + +## Live stop convergence + +`stopping` is never moved back to an active phase (`starting`/`streaming`/ +`reconnecting`). Any transition that did so would make the reconciler see an +active session again and publish another `video.stop_stream`, producing an +unbounded stop loop when the edge no longer knows the session. + +If the edge restarts `dock-video-relay`, it returns to `idle` and rejects the +stop with `VIDEO_SESSION_MISMATCH` (protocol section 5.2.4; the mock uses +`SESSION_NOT_FOUND`). Both codes mark the session `stopped` immediately. +Otherwise the session stays `stopping` and is forced to `stopped` after 120 +seconds (`stopForceAfter`), including when the stop command times out because +the edge is offline. That bound also keeps +`uk_live_session_active_dock` released so the dock can start a new session. + +## Live session lifetime + +A live session's absolute deadline comes from `live.max-session-seconds` (falling +back to `live.auth-expire-seconds` when unset). It is sent to the edge as +`video.start_stream.expiresAt` and also bounds the signed push/play URLs, so it +cannot be extended mid-session: re-sending `video.start_stream` with a different +`expiresAt` changes the params digest and the edge rejects it with +`COMMAND_ID_CONFLICT` (protocol section 5.1/6). + +Idle shutdown is therefore handled by viewer leases, not by that deadline: with +no active lease the session is stopped after `live.stop-grace-seconds`, while a +viewer that keeps calling the heartbeat endpoint keeps the stream alive until +the absolute deadline. The production value is 2 hours (`max-session-seconds: +7200`); raise it if users need longer continuous viewing. A `stopping` session +is forced to `stopped` after 120 seconds even if the edge keeps reporting +`streaming=true`, so a stuck publisher cannot block the dock from starting a new +session. + +## Live billing bitrate + +Live billing now charges the average bitrate the edge reports in +`state/video.bitrateBps` (the protocol's `expectedBitrateBps`, 1 Mbps by +default) instead of the session peak `maxBitrateBps` (1.5 Mbps). Run +`sql/024_live_session_billing_bitrate.sql` before deploying; it adds +`live_session.billing_bitrate_bps`. The migration is repeatable and +`sql/001_schema.sql` already contains the column for new databases. Until the +edge reports a bitrate the charged value falls back to `max_bitrate_bps`. + +## Live start params rehearsal + +`video.start_stream` retries must resend the exact same `params` because the +edge compares a params digest for `commandId` idempotency. The push URL carries +a short-lived credential and is therefore stored in Redis under +`laic:live:startparams:{commandId}` (never in MySQL, per the live backend +design, section 11) and replayed verbatim on retry. + ## Test traffic grant For QA environments, set the following configuration explicitly (it defaults diff --git a/cache/cache_key.go b/cache/cache_key.go index e5e97b8..40a3fd6 100644 --- a/cache/cache_key.go +++ b/cache/cache_key.go @@ -23,8 +23,9 @@ const ( TrafficKey = "laic:traffic:%d" // 用户云媒体流量余额（字节，Redis 为权威） - TrafficKeyPrefix = "laic:traffic:" // 流量余额 key 前缀（快照刷新扫描用） - LiveBilledKey = "laic:live:billed:%s" // 直播会话上次计费时间戳（sessionId） + TrafficKeyPrefix = "laic:traffic:" // 流量余额 key 前缀（快照刷新扫描用） + LiveBilledKey = "laic:live:billed:%s" // 直播会话上次计费时间戳（sessionId） + LiveStartParamsKey = "laic:live:startparams:%d" // video.start_stream 原始 params 缓存（commandId） ) func UserTokenKeyOf(userID int64, shortID string) string { @@ -90,3 +91,9 @@ func TrafficKeyOf(userID int64) string { func LiveBilledKeyOf(sessionID string) string { return fmt.Sprintf(LiveBilledKey, sessionID) } + +// LiveStartParamsKeyOf 缓存 video.start_stream 的完整 params。推流地址含短期鉴权串， +// 按设计不落 MySQL，但超时重试必须原样重发同一份 params。 +func LiveStartParamsKeyOf(commandID int64) string { + return fmt.Sprintf(LiveStartParamsKey, commandID) +} diff --git a/common/busi_error.go b/common/busi_error.go index 36d4380..4f35bc3 100644 --- a/common/busi_error.go +++ b/common/busi_error.go @@ -62,6 +62,7 @@ const ( LiveProviderInvalid = 57005 LiveLeaseNotFound = 57006 LiveStopping = 57007 + LiveNotReady = 57008 // 计费 58xxx TrafficNotEnough = 58001 @@ -124,6 +125,7 @@ var ( ErrLiveProviderInvalid = &BusiError{Code: LiveProviderInvalid, Msg: "直播服务配置无效"} ErrLiveLeaseNotFound = &BusiError{Code: LiveLeaseNotFound, Msg: "直播观看租约不存在或已过期"} ErrLiveStopping = &BusiError{Code: LiveStopping, Msg: "上一场直播正在停止，请稍后重试"} + ErrLiveNotReady = &BusiError{Code: LiveNotReady, Msg: "直播尚未就绪"} ErrVideoNotFound = &BusiError{Code: VideoNotFound, Msg: "视频不存在"} ErrVideoUploading = &BusiError{Code: VideoUploading, Msg: "视频上传中"} diff --git a/common/mysql.go b/common/mysql.go index 4a6ac42..d6c3e82 100644 --- a/common/mysql.go +++ b/common/mysql.go @@ -81,7 +81,13 @@ func extractConstraintName(errMsg string) string { start := strings.LastIndex(errMsg, "for key '") + len("for key '") end := strings.LastIndex(errMsg, "'") if start < end { - return errMsg[start:end] + name := errMsg[start:end] + // MySQL 8 的错误信息会带表名前缀（table.key），只保留索引名， + // 否则调用方按 'uk_xxx' 比较时永远匹配不到。 + if i := strings.LastIndex(name, "."); i >= 0 { + name = name[i+1:] + } + return name } return "" } diff --git a/config-prod.yaml b/config-prod.yaml index 101a953..051aa05 100644 --- a/config-prod.yaml +++ b/config-prod.yaml @@ -39,6 +39,9 @@ live: play-url-ttl-seconds: 300 protocol: srt srt-port: 1105 + # 单次直播会话的绝对时长上限，同时决定下发给设备的 expiresAt 与推流/播放 + # 地址签名有效期。会话中途无法延长（协议不允许重发不同 params 的 start_stream）， + # 而无人观看由观看租约 + stop-grace-seconds 控制。 max-session-seconds: 7200 viewer-lease-seconds: 30 stop-grace-seconds: 15 diff --git a/handler/live_handler.go b/handler/live_handler.go index 5a77d39..89db5d4 100644 --- a/handler/live_handler.go +++ b/handler/live_handler.go @@ -119,7 +119,14 @@ func StopLive(c *gin.Context) { common.FailWithBusiError(c, common.ErrParam) return } - if busiErr := service.DefaultLiveService.Stop(common.GetUserId(c), common.IsAdmin(c), dockID); busiErr != nil { + // streamSessionId 可在 body 或 query 中携带；为空时按兼容路径停止当前活动会话。 + var req vo.LiveStopReq + _ = c.ShouldBindJSON(&req) + sessionID := req.StreamSessionID + if sessionID == "" { + sessionID = c.Query("streamSessionId") + } + if busiErr := service.DefaultLiveService.Stop(common.GetUserId(c), common.IsAdmin(c), dockID, sessionID); busiErr != nil { common.FailWithBusiError(c, busiErr) return } diff --git a/model/media.go b/model/media.go index 57c4eb4..32389f0 100644 --- a/model/media.go +++ b/model/media.go @@ -11,6 +11,7 @@ type LiveSession struct { PushURLHash string `gorm:"column:push_url_hash;type:VARCHAR(128)" json:"pushUrlHash"` ExpiresAt int64 `gorm:"column:expires_at;type:BIGINT;not null" json:"expiresAt"` MaxBitrateBps int64 `gorm:"column:max_bitrate_bps;type:BIGINT;default:1500000" json:"maxBitrateBps"` + BillingBitrateBps int64 `gorm:"column:billing_bitrate_bps;type:BIGINT;not null;default:0" json:"billingBitrateBps"` Phase string `gorm:"column:phase;type:VARCHAR(16);default:idle" json:"phase"` StopReason string `gorm:"column:stop_reason;type:VARCHAR(32)" json:"stopReason"` ErrorCode string `gorm:"column:error_code;type:VARCHAR(64)" json:"errorCode"` diff --git a/mqtt/message.go b/mqtt/message.go index 338495f..4e71bb9 100644 --- a/mqtt/message.go +++ b/mqtt/message.go @@ -16,8 +16,6 @@ import ( // } type Envelope struct { RequestID string `json:"requestId"` - EventID string `json:"eventId"` - Version int64 `json:"version"` DockID string `json:"dockId"` DroneSN *string `json:"droneSn"` Timestamp int64 `json:"timestamp"` diff --git a/mqtt/mqtt.go b/mqtt/mqtt.go index 27b6d80..1da8fd0 100644 --- a/mqtt/mqtt.go +++ b/mqtt/mqtt.go @@ -2,6 +2,7 @@ package mqtt import ( "errors" + "regexp" mqtt "github.com/eclipse/paho.mqtt.golang" @@ -73,6 +74,15 @@ func inboundHandler(c mqtt.Client, msg mqtt.Message) { } } +// authKeyPattern 匹配推流/播放地址中的短期鉴权串。 +var authKeyPattern = regexp.MustCompile(`auth_key=[^"',&\s]+`) + +// redactSecrets 在下行 payload 落日志前遮蔽鉴权串。 +// 直播后台设计 §11 要求推流 URL 和播放鉴权参数不得进入日志。 +func redactSecrets(payload []byte) string { + return authKeyPattern.ReplaceAllString(string(payload), "auth_key=***") +} + // Publish 发布消息到指定 topic func Publish(topic string, qos byte, retained bool, payload []byte) error { if client == nil || !client.IsConnected() { @@ -82,7 +92,7 @@ func Publish(topic string, qos byte, retained bool, payload []byte) error { "topic="+topic, "qos=", qos, "retained=", retained, - "payload="+string(payload), + "payload="+redactSecrets(payload), ) token := client.Publish(topic, qos, retained, payload) token.Wait() diff --git a/service/billing_service.go b/service/billing_service.go index 3bbbd6d..6b38083 100644 --- a/service/billing_service.go +++ b/service/billing_service.go @@ -501,6 +501,18 @@ func mustID() int64 { return id } +// billingBitrateBps 返回计费用码率：优先使用设备上报的平均码率（协议 §7.8 +// 的 expectedBitrateBps），仅在设备尚未上报时退化到会话峰值码率。 +func billingBitrateBps(session *model.LiveSession) int64 { + if session.BillingBitrateBps > 0 { + return session.BillingBitrateBps + } + if session.MaxBitrateBps > 0 { + return session.MaxBitrateBps + } + return 1500000 +} + // chargeLiveSessions 只结算设备与云端均确认在线的 streaming 会话。 func (b *BillingService) chargeLiveSessions() { var sessions []model.LiveSession @@ -545,7 +557,7 @@ func (b *BillingService) chargeLiveSession(session *model.LiveSession, now time. if !now.After(periodStart) { return nil } - amount := int64(now.Sub(periodStart).Seconds() * float64(current.MaxBitrateBps) / 8) + amount := int64(now.Sub(periodStart).Seconds() * float64(billingBitrateBps(¤t)) / 8) if amount <= 0 { return nil } diff --git a/service/command_service.go b/service/command_service.go index 75ef1da..49695a0 100644 --- a/service/command_service.go +++ b/service/command_service.go @@ -166,6 +166,19 @@ func (s *CommandService) DispatchToDock(dockID, commandType string, params map[s auditParams["pushUrlHash"] = hex.EncodeToString(sum[:]) delete(auditParams, "pushUrl") } + if _, ok := params["pushUrl"].(string); ok { + logPushURL(dockID, params) + } + // 推流地址含短期鉴权串，按直播后台设计 §11 不能写入 MySQL；但协议 §5.1/§6 + // 要求超时重试复用同一 commandId 且 params 摘要完全一致，因此把原始 params + // 缓存在 Redis 供重试原样重发，避免重签名导致 COMMAND_ID_CONFLICT。 + if b, e := json.Marshal(params); e == nil { + if err := common.SetValueWithExpired(cache.LiveStartParamsKeyOf(commandID), string(b), liveStartParamsTTL()); err != nil { + logger.WARN("缓存推流启动参数失败", err) + } + } else { + logger.ERROR("序列化推流启动参数失败", e) + } } auditJSON, err := json.Marshal(auditParams) if err != nil { @@ -200,6 +213,61 @@ func cloneParams(params map[string]any) map[string]any { return copy } +// liveStartParamsTTL 返回推流启动参数缓存的有效期：覆盖整个会话时长再留出余量。 +func liveStartParamsTTL() uint { + seconds := int64(0) + if common.AppConf != nil { + seconds = common.AppConf.Live.MaxSessionSeconds + } + if seconds <= 0 { + seconds = 7200 + } + return uint(seconds + 600) +} + +// loadLiveStartParams 读取为指定 commandId 缓存的原始推流参数。 +func loadLiveStartParams(commandID int64) (map[string]any, bool) { + raw, err := common.GetBytes(cache.LiveStartParamsKeyOf(commandID)) + if err != nil || len(raw) == 0 { + return nil, false + } + var params map[string]any + if err := json.Unmarshal(raw, ¶ms); err != nil { + return nil, false + } + return params, true +} + +// logPushURL records the generated push URL so the edge allowlist rejection +// (VIDEO_PUSH_URL_NOT_ALLOWED) can be diagnosed. The signed auth_key is +// redacted so the credential never lands in logs; everything else (scheme, +// host, port, streamid) is kept intact for comparison with the edge allowlist. +func logPushURL(dockID string, params map[string]any) { + pushURL, _ := params["pushUrl"].(string) + streamSessionID, _ := params["streamSessionId"].(string) + logger.INFO("下发推流地址", + "dockId="+dockID, + "streamSessionId="+streamSessionID, + "len="+strconv.Itoa(len(pushURL)), + "url="+redactPushURL(pushURL)) +} + +// redactPushURL masks only the auth_key value, keeping the rest of the URL +// intact for comparison against the device-side allowlist. +func redactPushURL(pushURL string) string { + const marker = "auth_key=" + start := strings.Index(pushURL, marker) + if start < 0 { + return pushURL + } + valueStart := start + len(marker) + end := strings.IndexAny(pushURL[valueStart:], ",&") + if end < 0 { + return pushURL[:valueStart] + "***" + } + return pushURL[:valueStart] + "***" + pushURL[valueStart+end:] +} + func (s *CommandService) StartRetryScanner() { go func() { ticker := time.NewTicker(10 * time.Second) @@ -253,7 +321,20 @@ func (s *CommandService) retryTimeout() { _ = json.Unmarshal([]byte(cmd.Params), ¶ms) } if cmd.CommandType == "video.start_stream" { - params = DefaultLiveService.RefreshStartParams(cmd.DockID, params) + // 审计参数里只有 pushUrlHash；重试必须使用缓存的原样 params， + // 否则参数摘要变化会被工控机按 COMMAND_ID_CONFLICT 拒绝。 + sessionID, _ := params["streamSessionId"].(string) + cached, ok := loadLiveStartParams(cmd.ID) + if !ok { + logger.ERROR("重试缺少缓存的推流启动参数，跳过重发 commandId="+strconv.FormatInt(cmd.ID, 10), errors.New("live start params lost")) + if sessionID != "" { + _ = common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting')", sessionID, cmd.DockID). + Updates(map[string]any{"phase": "failed", "error_code": "START_PARAMS_LOST"}).Error + } + continue + } + params = cached + logPushURL(cmd.DockID, params) } b, _ := json.Marshal(s.commandEnvelope(&cmd, requestID, params)) topic := fmt.Sprintf("dock-edge/v1/dock/%s/command", cmd.DockID) diff --git a/service/live_service.go b/service/live_service.go index 5bdd96d..1d8063e 100644 --- a/service/live_service.go +++ b/service/live_service.go @@ -99,12 +99,10 @@ func (s *LiveService) Join(userID int64, isAdmin bool, dockID string, req *vo.Li if err != nil { return err } - expireSec := common.AppConf.Live.AuthExpireSeconds - if expireSec <= 0 { - expireSec = 7200 - } - expiresAt := time.Now().Add(time.Duration(expireSec) * time.Second) + expiresAt := sessionExpiry(time.Now()) streamName := strconv.FormatInt(streamID, 10) + // pushUrl 的签名有效期与设备硬停止时间保持一致，保证整个会话期间 + // 重试下发的同一个 pushUrl 仍然有效（协议 §7.2）。 credentials, err := provider.CreateStream(liveprovider.StreamRequest{StreamName: streamName, ExpiresAt: expiresAt, MaxBitrateBps: maxBitrate}) if err != nil { return err @@ -211,6 +209,34 @@ func stopGraceSeconds() time.Duration { return time.Duration(seconds) * time.Second } +// stopForceAfter 是 stopping 的强制收敛期限。设备可能因为推流子进程卡死而一直 +// 上报 streaming=true，必须在有限时间内把会话判定为 stopped，否则 +// uk_live_session_active_dock 会让该机巢再也无法开启新直播。 +const stopForceAfter = 120 * time.Second + +func stoppingDeadline() time.Time { + return time.Now().Add(stopForceAfter) +} + +// sessionExpiry 返回会话的绝对过期时间。它有两个用途： +// 1. 作为 video.start_stream 的 expiresAt 下发给设备（设备侧硬停止时间）； +// 2. 作为推流/播放地址的签名有效期上限。 +// +// 协议不允许在会话中途延长 expiresAt（重发 start_stream 会因参数摘要变化被 +// 按 COMMAND_ID_CONFLICT 拒绝），所以这里必须在创建时一次性给足。 +// “无人观看就停”由观看租约 + 宽限期控制，会话在有人持续心跳时会一直保持到 +// 这个上限。取值以 max-session-seconds 为准，未配置时退回 auth-expire-seconds。 +func sessionExpiry(now time.Time) time.Time { + seconds := common.AppConf.Live.MaxSessionSeconds + if seconds <= 0 { + seconds = common.AppConf.Live.AuthExpireSeconds + } + if seconds <= 0 { + seconds = 7200 + } + return now.Add(time.Duration(seconds) * time.Second) +} + // ReconcileLeases expires stale viewer leases and stops unviewed sessions after the grace period. func (s *LiveService) ReconcileLeases() { now := time.Now() @@ -288,14 +314,20 @@ func (s *LiveService) reconcileCloudSession(provider liveprovider.Adapter, sessi case "starting", "reconnecting": if session.DeviceStreaming && online.Online { updates["phase"] = "streaming" - updates["started_at"] = now + // 只在首次进入 streaming 时记开始时间，避免重连后把计费周期起点重置 + if session.StartedAt == nil { + updates["started_at"] = now + } } case "streaming": if !session.DeviceStreaming || !online.Online { updates["phase"] = "reconnecting" } case "stopping": - if !session.DeviceStreaming && !online.Online { + // 设备与云端都离线，或超过强制收敛期限 → 判定已停止。 + // 后者避免设备（推流子进程）卡死时会话永久停在 stopping。 + if (!session.DeviceStreaming && !online.Online) || + (session.StopDeadline != nil && now.After(*session.StopDeadline)) { updates["phase"] = "stopped" updates["stopped_at"] = now } @@ -303,29 +335,32 @@ func (s *LiveService) reconcileCloudSession(provider liveprovider.Adapter, sessi _ = common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND phase = ?", session.ID, session.DockID, session.Phase).Updates(updates).Error } -func (s *LiveService) RefreshStartParams(dockID string, params map[string]any) map[string]any { - if params == nil { - return params - } - streamSessionID, _ := params["streamSessionId"].(string) - if streamSessionID == "" { - return params - } - var session model.LiveSession - if common.DB.Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", streamSessionID, dockID).First(&session).Error != nil { - return params - } - provider, busiErr := s.adapter() - if busiErr != nil { - return params +// OnDeviceIdle 处理设备自述“没有会话且不在推流”（state/video 里 streamSessionId 为空 +// 且 streaming=false）。dock-video-relay 只把会话存在进程内存，重启后会回到 idle， +// 此时继续等阿里云超时只会让会话卡在 reconnecting。 +// +// 只对“已确认过”的会话生效（streaming/reconnecting）：starting 阶段的设备还没收到 +// start_stream，上报的 idle 属于正常状态，不能误杀。 +func (s *LiveService) OnDeviceIdle(dockID string) { + now := time.Now() + result := common.DB.Model(&model.LiveSession{}). + Where("dock_id = ? AND phase IN ('streaming','reconnecting')", dockID). + Updates(map[string]any{ + "phase": "stopped", + "stop_reason": "device_restarted", + "error_code": "DEVICE_SESSION_LOST", + "device_streaming": false, + "cloud_online": false, + "stopped_at": now, + "updated_at": now, + }) + if result.Error != nil { + logger.ERROR("处理设备空闲状态失败", result.Error) + return } - credentials, err := provider.CreateStream(liveprovider.StreamRequest{StreamName: session.StreamName, ExpiresAt: time.Unix(session.ExpiresAt, 0), MaxBitrateBps: session.MaxBitrateBps}) - if err != nil { - return params + if result.RowsAffected > 0 { + logger.WARN("设备已无直播会话，会话判定结束", dockID, result.RowsAffected) } - params["pushUrl"] = credentials.PushURL - params["expiresAt"] = credentials.ExpiresAt.UnixMilli() - return params } func (s *LiveService) OnCommandTimeout(cmd model.DeviceCommandLog) { @@ -335,13 +370,22 @@ func (s *LiveService) OnCommandTimeout(cmd model.DeviceCommandLog) { if json.Unmarshal([]byte(cmd.Params), ¶ms) != nil || params.StreamSessionID == "" { return } - updates := map[string]any{"error_code": "COMMAND_ACK_TIMEOUT"} if cmd.CommandType == "video.start_stream" { - updates["phase"] = "failed" - } else { - updates["phase"] = "reconnecting" + _ = common.DB.Model(&model.LiveSession{}). + Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting')", params.StreamSessionID, cmd.DockID). + Updates(map[string]any{"phase": "failed", "error_code": "COMMAND_ACK_TIMEOUT"}).Error + return } - _ = common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", params.StreamSessionID, cmd.DockID).Updates(updates).Error + // 停止指令超时：绝不能再把会话改回活跃态（reconnecting），否则对账会重新 + // 扫描到活跃会话并再次下发停止指令，形成“停→复活→再停”的死循环。 + // 这里保持 stopping，由 stop_deadline 在有限时间内强制收敛为 stopped； + // COALESCE 保证已设置的期限不被重复超时向后推。 + _ = common.DB.Model(&model.LiveSession{}). + Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", params.StreamSessionID, cmd.DockID). + Updates(map[string]any{ + "error_code": "STOP_COMMAND_ACK_TIMEOUT", + "stop_deadline": gorm.Expr("COALESCE(stop_deadline, ?)", stoppingDeadline()), + }).Error } // Start 兼容旧启动接口，统一复用创建或加入会话的租约流程。 @@ -353,8 +397,9 @@ func (s *LiveService) Start(userID int64, isAdmin bool, dockID string, req *vo.L return result.Session, nil } -// Stop 停止指定机巢最新活动会话；设备必须根据 streamSessionId 防止误停新会话。 -func (s *LiveService) Stop(userID int64, isAdmin bool, dockID string) *common.BusiError { +// Stop 停止指定机巢的活动会话。sessionID 非空时必须与当前活动会话一致， +// 避免旧页面拿着已过期的会话号把新会话停掉；设备仍须根据 streamSessionId 防误停。 +func (s *LiveService) Stop(userID int64, isAdmin bool, dockID, sessionID string) *common.BusiError { var dock model.Dock if err := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("dock_id = ?", dockID).First(&dock).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { @@ -369,15 +414,33 @@ func (s *LiveService) Stop(userID int64, isAdmin bool, dockID string) *common.Bu } return common.ErrInternal } + // 调用方带了会话号就必须匹配当前活动会话，防止旧页面/旧任务停掉新会话。 + if sessionID != "" && sessionID != active.ID { + logger.WARN("拒绝停止非当前直播会话", dockID, sessionID, active.ID) + return common.NewBusiError(common.LiveNotFound, "该直播会话已结束或已更换") + } + if sessionID == "" { + logger.WARN("停止直播未携带 streamSessionId（旧客户端兼容路径）", dockID) + } + // 先做条件置位抢占：只有把会话从活跃态推到 stopping 的调用者才下发停流指令。 + // 重复调用（前端重试、多人同时点停止、对账任务并发）只会幂等返回，不会重复下发。 + result := common.DB.Model(&model.LiveSession{}). + Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting')", active.ID, dockID). + Updates(map[string]any{"phase": "stopping", "stop_reason": "manual", "stop_deadline": stoppingDeadline()}) + if result.Error != nil { + return common.ErrInternal + } + if result.RowsAffected == 0 { + return nil + } params := map[string]any{ "streamSessionId": active.ID, "reason": "user_request", } if _, err := DefaultCommandService.DispatchToDock(dockID, "video.stop_stream", params); err != nil { logger.ERROR("下发停止推流指令失败", err) - return common.ErrInternal - } - if err := common.DB.Model(&model.LiveSession{}).Where("id = ? AND phase IN ('starting','streaming','reconnecting')", active.ID).Updates(map[string]any{"phase": "stopping", "stop_reason": "manual"}).Error; err != nil { + _ = common.DB.Model(&model.LiveSession{}).Where("id = ? AND phase = 'stopping'", active.ID). + Updates(map[string]any{"error_code": "STOP_COMMAND_PUBLISH_FAILED"}).Error return common.ErrInternal } return nil @@ -390,7 +453,7 @@ func (s *LiveService) stopForBilling(dockID, streamSessionID string, reason stri // Transition first with a conditional update. This is the idempotency gate // for concurrent billing/reconciliation workers: only the worker that moves // the session into stopping is allowed to publish video.stop_stream. - updates := map[string]any{"phase": "stopping", "stop_reason": reason} + updates := map[string]any{"phase": "stopping", "stop_reason": reason, "stop_deadline": stoppingDeadline()} if reason == "no_balance" { updates["error_code"] = "TRAFFIC_NOT_ENOUGH" } @@ -405,7 +468,7 @@ func (s *LiveService) stopForBilling(dockID, streamSessionID string, reason stri } params := map[string]any{ "streamSessionId": streamSessionID, - "reason": reason, + "reason": protocolStopReason(reason), } cmd, err := DefaultCommandService.DispatchToDock(dockID, "video.stop_stream", params) if err != nil { @@ -437,7 +500,9 @@ func (s *LiveService) GetPlayURL(userID int64, isAdmin bool, dockID, sessionID s return nil, busiErr } var session model.LiveSession - query := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("dock_id = ? AND phase = 'streaming' AND (expires_at = 0 OR expires_at > ?)", dockID, time.Now().Unix()) + query := common.DB.Model(&model.LiveSession{}). + Scopes(withDockFilter(userID, isAdmin)). + Where("dock_id = ? AND phase = 'streaming' AND (expires_at = 0 OR expires_at > ?)", dockID, time.Now().Unix()) if sessionID != "" { query = query.Where("id = ?", sessionID) } else { @@ -449,12 +514,31 @@ func (s *LiveService) GetPlayURL(userID int64, isAdmin bool, dockID, sessionID s WHERE viewer_id = ? AND released_at IS NULL AND expires_at > ? )`, userID, time.Now()).Order("created_at DESC, id DESC") } - if err := query.First(&session).Error; err != nil { - if errors.Is(err, gorm.ErrRecordNotFound) { - return nil, common.ErrLiveNotFound - } + // 用 Find 而不是 First：会话尚未进入 streaming 属于正常等待态， + // 不应该被 GORM 记成 record not found 的 ERROR 日志。 + result := query.Limit(1).Find(&session) + if result.Error != nil { + logger.ERROR("查询直播会话失败", result.Error) return nil, common.ErrInternal } + if result.RowsAffected == 0 { + // 区分“还没就绪”和“已经结束”，让前端知道该继续等还是该收尾。 + var active int64 + existQuery := common.DB.Model(&model.LiveSession{}). + Scopes(withDockFilter(userID, isAdmin)). + Where("dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", dockID) + if sessionID != "" { + existQuery = existQuery.Where("id = ?", sessionID) + } + if err := existQuery.Count(&active).Error; err != nil { + logger.ERROR("查询直播会话状态失败", err) + return nil, common.ErrInternal + } + if active > 0 { + return nil, common.ErrLiveNotReady + } + return nil, common.ErrLiveNotFound + } var lease model.LiveViewerLease if err := common.DB.Where("stream_session_id = ? AND viewer_id = ? AND released_at IS NULL AND expires_at > ?", session.ID, userID, time.Now()).First(&lease).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { @@ -470,7 +554,16 @@ func (s *LiveService) GetPlayURL(userID int64, isAdmin bool, dockID, sessionID s if err != nil { return nil, common.NewBusiError(common.LiveProviderInvalid, "生成播放地址失败") } - return &vo.LivePlayURLVO{StreamName: session.StreamName, PlayURL: urls.HLS, ExpiresAt: urls.ExpiresAt.Unix()}, nil + // 同时返回 HLS / FLV / RTMP：HLS 兼容性最好但延迟高，实时监控用 FLV + // （HTTP-FLV）延迟可压到 1~3 秒。playUrl 仍保持 HLS，避免破坏旧前端。 + return &vo.LivePlayURLVO{ + StreamName: session.StreamName, + PlayURL: urls.HLS, + PlayURLHLS: urls.HLS, + PlayURLFLV: urls.FLV, + PlayURLRTMP: urls.RTMP, + ExpiresAt: urls.ExpiresAt.Unix(), + }, nil } func (s *LiveService) OnCommandAck(cmd *model.DeviceCommandLog, accepted bool, resultCode string) { @@ -493,18 +586,19 @@ func (s *LiveService) OnCommandAck(cmd *model.DeviceCommandLog, accepted bool, r if cmd.CommandType == "video.start_stream" { updates["phase"] = "failed" } else { - // A stop rejection means the stream is still active for ordinary - // command failures. SESSION_NOT_FOUND is different: the edge has - // already discarded this session (typically after a mock restart), so - // restoring streaming would make expiry reconciliation send stop forever. - if resultCode == "SESSION_NOT_FOUND" || params.Reason == "session_expired" { + // 设备已经不认这个会话（工控机/视频服务重启后回到 idle，或会话已被替换）： + // 继续重试停止毫无意义，直接判定结束，释放活动会话唯一约束。 + // SESSION_NOT_FOUND 是 mock 的码，VIDEO_SESSION_MISMATCH 是协议 §5.2.4 的码。 + if resultCode == "SESSION_NOT_FOUND" || resultCode == "VIDEO_SESSION_MISMATCH" { now := time.Now() updates["phase"] = "stopped" updates["device_streaming"] = false updates["cloud_online"] = false updates["stopped_at"] = now } else { - updates["phase"] = "streaming" + // 一般停止失败（设备仍在推流等）保持 stopping，由 stop_deadline 强制收敛。 + // 绝不能改回 streaming，否则会与停流对账形成死循环。 + updates["phase"] = "stopping" } } if err := common.DB.Model(&model.LiveSession{}). @@ -514,42 +608,68 @@ func (s *LiveService) OnCommandAck(cmd *model.DeviceCommandLog, accepted bool, r } } -// OnVideoState accepts only ordered state from the matching non-terminal stream session. -func (s *LiveService) OnVideoState(dockID string, streaming bool, streamSessionID, eventID string, version, updatedAt int64) bool { - if streamSessionID == "" || eventID == "" || version <= 0 || updatedAt <= 0 { +// OnVideoState accepts ordered state from the matching non-terminal stream session. +// sequence 是幂等去重用的单调序号（协议无 version，由 state/video 快照的 +// updatedAt 毫秒充当），eventId 是派生的 -， +// 两者仍旧落在 device_state_version / device_event_id 两个既有列上以兼容历史数据。 +func (s *LiveService) OnVideoState(dockID string, streaming bool, streamSessionID, eventID string, sequence, updatedAt, bitrateBps int64) bool { + if streamSessionID == "" || eventID == "" || sequence <= 0 || updatedAt <= 0 { + logger.WARN("视频状态缺少顺序元数据（OnVideoState）", dockID, streamSessionID, eventID, sequence, updatedAt) return false } now := time.Now() if abs(now.UnixMilli()-updatedAt) > int64((5*time.Minute)/time.Millisecond) { - logger.WARN("忽略过期视频状态", dockID, streamSessionID) + // updatedAt 单位不一致（秒 vs 毫秒）或设备时钟偏差会走到这里。 + logger.WARN("忽略过期视频状态", dockID, streamSessionID, "nowMs", now.UnixMilli(), "updatedAt", updatedAt) return false } var session model.LiveSession if err := common.DB.Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping')", streamSessionID, dockID).First(&session).Error; err != nil { + logger.WARN("视频状态对应的直播会话不存在或已结束", dockID, streamSessionID, err) return false } - if version <= session.DeviceStateVersion || (session.DeviceEventID != "" && session.DeviceEventID == eventID) || updatedAt < session.DeviceUpdatedAt { + // 重复投递由序号是否前进判定；eventId 只用于“跨会话归属”校验， + // 因为协议 §7.8 的 state/video 是快照，requestId 可能在同一会话内被复用。 + if sequence <= session.DeviceStateVersion || updatedAt < session.DeviceUpdatedAt { + logger.WARN("忽略乱序或重复的视频状态", dockID, streamSessionID, + "sequence", sequence, "curSequence", session.DeviceStateVersion, + "eventId", eventID, "curEventId", session.DeviceEventID, + "updatedAt", updatedAt, "curUpdatedAt", session.DeviceUpdatedAt) return false } - // eventId is expected to be globally unique. Reject a stale/reused device - // event that was already recorded for another live session before touching - // the unique device-event column. + // eventId 由 - 派生，天然唯一；这里只防止同一事件被 + // 陈旧会话重复占用（device_event_id 有唯一索引 uk_live_session_device_event）。 var eventOwner model.LiveSession if result := common.DB.Select("id").Where("device_event_id = ? AND id <> ?", eventID, streamSessionID).Limit(1).Find(&eventOwner); result.Error != nil { + logger.ERROR("校验视频状态事件唯一性失败", result.Error) return false } else if result.RowsAffected > 0 { logger.WARN("忽略已被其他直播会话使用的视频状态事件", eventID, streamSessionID, eventOwner.ID) return false } updates := map[string]any{ - "device_state_version": version, + "device_state_version": sequence, "device_event_id": eventID, "device_updated_at": updatedAt, "device_streaming": streaming, "updated_at": now, } - if streaming { + // 协议 §7.8：bitrateBps 是设备侧 expectedBitrateBps（平均码率），作为计费口径。 + // 只在推流且有有效值时写入，避免停止时把计费码率清零。 + if streaming && bitrateBps > 0 { + updates["billing_bitrate_bps"] = bitrateBps + } + if session.Phase == "stopping" { + // 已下发扬停指令：设备可能仍在上报 streaming=true（或保留了停止后的会话号）， + // 绝不能把 phase 改回 streaming，否则会和停流对账来回拉扯成死循环。 + // 这里只记录设备侧事实，phase 交给对账任务收敛到 stopped。 + if streaming { + updates["error_code"] = "STOP_IN_PROGRESS" + } else { + updates["error_code"] = "DEVICE_STREAM_OFFLINE" + } + } else if streaming { provider, busiErr := s.adapter() if busiErr != nil { return false @@ -557,25 +677,23 @@ func (s *LiveService) OnVideoState(dockID string, streaming bool, streamSessionI online, queryErr := provider.QueryOnline(session.StreamName) if queryErr != nil || !online.Online { updates["error_code"] = "PROVIDER_ONLINE_UNVERIFIED" - result := common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND device_state_version < ?", streamSessionID, dockID, version).Updates(updates) + result := common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND device_state_version < ?", streamSessionID, dockID, sequence).Updates(updates) if result.Error != nil { - if code, constraint := common.ParseError(result.Error); code == 1062 && constraint == "uk_live_session_device_event" { - return false - } return false } return true } updates["phase"] = "streaming" - updates["started_at"] = now + // 只在首次进入 streaming 时记开始时间，避免重连后把计费周期起点重置 + if session.StartedAt == nil { + updates["started_at"] = now + } updates["cloud_confirmed_at"] = now - } else if session.Phase == "stopping" { - updates["error_code"] = "DEVICE_STREAM_OFFLINE" } else { updates["phase"] = "reconnecting" updates["error_code"] = "DEVICE_STREAM_OFFLINE" } - result := common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping') AND device_state_version < ?", streamSessionID, dockID, version).Updates(updates) + result := common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting','stopping') AND device_state_version < ?", streamSessionID, dockID, sequence).Updates(updates) if result.Error != nil { if code, constraint := common.ParseError(result.Error); code == 1062 && constraint == "uk_live_session_device_event" { return false @@ -591,3 +709,14 @@ func abs(value int64) int64 { } return value } + +// protocolStopReason 把内部停止原因映射为协议 §5.2.4 允许的取值， +// 避免工控机按 INVALID_PARAM 拒绝停止命令；真实原因仍记录在 live_session.stop_reason。 +func protocolStopReason(reason string) string { + switch reason { + case "no_viewers", "user_request", "admin_stop", "session_replaced": + return reason + default: + return "user_request" + } +} diff --git a/service/mqtt_subscriber.go b/service/mqtt_subscriber.go index f94be32..2e407fd 100644 --- a/service/mqtt_subscriber.go +++ b/service/mqtt_subscriber.go @@ -93,9 +93,9 @@ type otaReported struct { } // videoState §7.8 视频推流状态 +// 协议外层（§4）和 payload 都没有 eventId/version，只有 updatedAt（毫秒）， +// 因为 state/video 是状态快照而不是事件流。 type videoState struct { - Version int64 `json:"version"` - EventID string `json:"eventId"` Provider string `json:"provider"` Phase string `json:"phase"` InputOnline bool `json:"inputOnline"` @@ -106,7 +106,7 @@ type videoState struct { Width *int `json:"width"` Height *int `json:"height"` FrameRate *float64 `json:"frameRate"` - BitrateBps uint64 `json:"bitrateBps"` + BitrateBps int64 `json:"bitrateBps"` RetryCount int `json:"retryCount"` StopReason *string `json:"stopReason"` ErrorCode *string `json:"errorCode"` @@ -291,20 +291,40 @@ func handleVideo(dockID string, env *mqtt.Envelope) { _ = json.Unmarshal(env.Payload, &raw) var st videoState if err := json.Unmarshal(env.Payload, &st); err != nil { - logger.ERROR("解析 state/video 失败", err) + logger.ERROR("解析 state/video 失败 payload="+string(env.Payload), err) return } - if st.EventID == "" { - st.EventID = env.EventID + // mqtt_interface_design.md §7.8：state/video 是保留状态快照，协议外层（§4） + // 和 payload 都没有 eventId/version（“状态快照不需要单独的事件 ID”），只有 + // updatedAt（毫秒）。因此用 updatedAt 充当单调序号，并以 + // - 作为幂等去重键（内部仍存入 device_event_id）。 + sequence := st.UpdatedAt + eventID := "" + if st.StreamSessionID != "" && st.UpdatedAt > 0 { + eventID = st.StreamSessionID + "-" + strconv.FormatInt(st.UpdatedAt, 10) } - if st.Version == 0 { - st.Version = env.Version - } - if st.StreamSessionID == "" || st.EventID == "" || st.Version <= 0 || st.UpdatedAt <= 0 { - logger.WARN("忽略缺少顺序元数据的视频状态", dockID, st.StreamSessionID) + // 无条件记录每一条 state/video，用于判断设备是否真的在上报。 + logger.INFO("收到视频状态", + "dockId="+dockID, + "requestId="+env.RequestID, + "streamSessionId="+st.StreamSessionID, + "streaming="+strconv.FormatBool(st.Streaming), + "phase="+st.Phase, + "updatedAt="+strconv.FormatInt(st.UpdatedAt, 10), + "eventId="+eventID, + "raw="+string(env.Payload)) + if st.StreamSessionID == "" || eventID == "" || sequence <= 0 || st.UpdatedAt <= 0 { + // 设备自述“没有会话且不在推流”是权威的重启信号：dock-video-relay 把会话 + // 保存在进程内存里，重启后会回到 idle + streamSessionId=null。这条消息 + // 之前被直接丢弃，导致后端会话只能卡在 reconnecting 直到 expires_at。 + if st.StreamSessionID == "" && !st.Streaming { + DefaultLiveService.OnDeviceIdle(dockID) + } + logger.WARN("忽略缺少顺序元数据的视频状态", dockID, st.StreamSessionID, string(env.Payload)) return } - if !DefaultLiveService.OnVideoState(dockID, st.Streaming, st.StreamSessionID, st.EventID, st.Version, st.UpdatedAt) { + if !DefaultLiveService.OnVideoState(dockID, st.Streaming, st.StreamSessionID, eventID, sequence, st.UpdatedAt, st.BitrateBps) { + logger.WARN("视频状态被顺序或归属校验拒绝", dockID, st.StreamSessionID, eventID, sequence, st.UpdatedAt) return } _ = common.HashSetValues(cache.VideoKeyOf(dockID), flatten(raw)) diff --git a/sql/001_schema.sql b/sql/001_schema.sql index 360428a..ab5f17c 100644 --- a/sql/001_schema.sql +++ b/sql/001_schema.sql @@ -217,6 +217,7 @@ CREATE TABLE IF NOT EXISTS live_session ( push_url_hash VARCHAR(128) COMMENT '推流地址哈希', expires_at BIGINT NOT NULL COMMENT '过期时间戳', max_bitrate_bps BIGINT DEFAULT 1500000 COMMENT '最大码率（比特/秒）', + billing_bitrate_bps BIGINT NOT NULL DEFAULT 0 COMMENT '设备上报的平均码率（bit/s），计费口径', phase VARCHAR(16) DEFAULT 'idle' COMMENT '直播会话阶段', stop_reason VARCHAR(32) COMMENT '停止原因', error_code VARCHAR(64) COMMENT '错误码', @@ -236,9 +237,11 @@ CREATE TABLE IF NOT EXISTS live_session ( active_dock_id VARCHAR(64) GENERATED ALWAYS AS ( CASE WHEN phase IN ('starting', 'streaming', 'reconnecting', 'stopping') THEN dock_id ELSE NULL END ) STORED COMMENT '活动机巢标识（生成列）', + device_event_key VARCHAR(128) GENERATED ALWAYS AS (NULLIF(device_event_id, '')) STORED + COMMENT '设备事件唯一键（空串转 NULL，允许多条未上报事件的会话）', KEY idx_live_session_dock_phase (dock_id, phase, created_at), KEY idx_live_session_phase_deadline (phase, stop_deadline), - UNIQUE KEY uk_live_session_device_event (device_event_id), + UNIQUE KEY uk_live_session_device_event (device_event_key), UNIQUE KEY uk_live_session_active_dock (active_dock_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/vo/media_vo.go b/vo/media_vo.go index d6cf1f6..d4a1873 100644 --- a/vo/media_vo.go +++ b/vo/media_vo.go @@ -18,6 +18,12 @@ type LiveStartReq struct { MaxBitrateBps int64 `json:"maxBitrateBps"` } +// LiveStopReq 停止直播。streamSessionId 建议携带：省略时为兼容旧客户端的 +// “停止当前活动会话”，携带时必须与当前活动会话一致，避免旧页面停掉新会话。 +type LiveStopReq struct { + StreamSessionID string `json:"streamSessionId"` +} + // LiveSessionCreateReq 创建或加入指定机巢的活动直播会话。 type LiveSessionCreateReq struct { MaxBitrateBps int64 `json:"maxBitrateBps"` @@ -29,11 +35,15 @@ type LiveSessionVO struct { LeaseExpiresAt int64 `json:"leaseExpiresAt"` } -// LivePlayURLVO 直播播放地址 +// LivePlayURLVO 直播播放地址。playUrl 保留 HLS 以兼容旧前端； +// 实时监控建议用 playUrlFlv（mpegts.js），延迟远低于 HLS。 type LivePlayURLVO struct { - StreamName string `json:"streamName"` - PlayURL string `json:"playUrl"` - ExpiresAt int64 `json:"expiresAt"` + StreamName string `json:"streamName"` + PlayURL string `json:"playUrl"` + PlayURLHLS string `json:"playUrlHls"` + PlayURLFLV string `json:"playUrlFlv"` + PlayURLRTMP string `json:"playUrlRtmp"` + ExpiresAt int64 `json:"expiresAt"` } // VideoUploadReq 申请视频上传