From 77c224696c938473d68f8e71c70307dcc0e30fe6 Mon Sep 17 00:00:00 2001 From: liuhaodong Date: Wed, 2 Sep 2026 17:29:28 +0800 Subject: [PATCH] =?UTF-8?q?feat=EF=BC=9A=E5=8A=9F=E8=83=BD=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- live/providers.go | 21 +++++++++--- service/billing_service.go | 13 +++----- service/live_service.go | 66 +++++++++++++++++++++++++++++--------- 3 files changed, 73 insertions(+), 27 deletions(-) diff --git a/live/providers.go b/live/providers.go index cd37ed4..08dd33e 100644 --- a/live/providers.go +++ b/live/providers.go @@ -75,10 +75,23 @@ func (a AliyunAdapter) CreateStream(req StreamRequest) (StreamCredentials, error func (a AliyunAdapter) CreatePlayURLs(req PlayRequest) (PlayURLs, error) { expires := req.ExpiresAt path := "/" + a.cfg.AppName + "/" + req.StreamName - signature := sign(path, expires.Unix(), a.cfg.PlayAuthKey) - base := "https://" + a.cfg.PlayDomain + "/" + a.cfg.AppName + "/" + req.StreamName - rtmp := "rtmp://" + a.cfg.PlayDomain + "/" + a.cfg.AppName + "/" + req.StreamName + "?auth_key=" + signature - return PlayURLs{HLS: base + ".m3u8?auth_key=" + signature, FLV: base + ".flv?auth_key=" + signature, RTMP: rtmp, ExpiresAt: expires}, nil + // Alibaba Live authenticates the complete playback path. HLS and FLV + // therefore need independent signatures because their file extensions are + // part of the URI used by the CDN auth check. + hlsPath := path + ".m3u8" + flvPath := path + ".flv" + hlsSignature := sign(hlsPath, expires.Unix(), a.cfg.PlayAuthKey) + flvSignature := sign(flvPath, expires.Unix(), a.cfg.PlayAuthKey) + base := "http://" + a.cfg.PlayDomain + "/" + a.cfg.AppName + "/" + req.StreamName + rtmpPath := path + rtmpSignature := sign(rtmpPath, expires.Unix(), a.cfg.PlayAuthKey) + rtmp := "rtmp://" + a.cfg.PlayDomain + "/" + a.cfg.AppName + "/" + req.StreamName + "?auth_key=" + rtmpSignature + return PlayURLs{ + HLS: base + ".m3u8?auth_key=" + hlsSignature, + FLV: base + ".flv?auth_key=" + flvSignature, + RTMP: rtmp, + ExpiresAt: expires, + }, nil } func (a AliyunAdapter) QueryOnline(streamName string) (OnlineStatus, error) { return a.client.queryOnline(streamName) diff --git a/service/billing_service.go b/service/billing_service.go index f77d52b..d2e3531 100644 --- a/service/billing_service.go +++ b/service/billing_service.go @@ -571,13 +571,10 @@ func (b *BillingService) chargeLiveSession(session *model.LiveSession, now time. if err := tx.Model(&segment).Update("status", "insufficient").Error; err != nil { return err } - result := tx.Model(&model.LiveSession{}).Where("id = ? AND phase = 'streaming'", current.ID).Updates(map[string]any{ - "phase": "stopping", "stop_reason": "no_balance", "error_code": "TRAFFIC_NOT_ENOUGH", "updated_at": now, - }) - if result.Error != nil { - return result.Error - } - needStop = result.RowsAffected == 1 + // stopForBilling performs the conditional streaming -> stopping + // transition after this transaction commits, and acts as the single + // publish gate for concurrent workers. + needStop = true return nil } @@ -613,7 +610,7 @@ func (b *BillingService) chargeLiveSession(session *model.LiveSession, now time. return } if needStop { - if _, busiErr := DefaultLiveService.stopForBilling(session.DockID, session.ID); busiErr != nil { + if _, busiErr := DefaultLiveService.stopForBilling(session.DockID, session.ID, "no_balance"); busiErr != nil { logger.ERROR("直播余额不足停止推流失败", busiErr) } return diff --git a/service/live_service.go b/service/live_service.go index be2dcea..6da1845 100644 --- a/service/live_service.go +++ b/service/live_service.go @@ -152,16 +152,22 @@ func (s *LiveService) GetSession(userID int64, isAdmin bool, dockID, streamSessi } func (s *LiveService) Heartbeat(userID int64, isAdmin bool, dockID, streamSessionID string) (*vo.LiveSessionVO, *common.BusiError) { var session model.LiveSession - if err := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("id = ? AND dock_id = ?", streamSessionID, dockID).First(&session).Error; err != nil { + if err := common.DB.Scopes(withDockFilter(userID, isAdmin)).Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting') AND (expires_at = 0 OR expires_at > ?)", streamSessionID, dockID, time.Now().Unix()).First(&session).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil, common.ErrLiveNotFound } return nil, common.ErrInternal } expiresAt := time.Now().Add(time.Duration(leaseSeconds()) * time.Second) - result := common.DB.Model(&model.LiveViewerLease{}).Where("stream_session_id = ? AND viewer_id = ? AND released_at IS NULL", streamSessionID, userID).Updates(map[string]any{"expires_at": expiresAt, "updated_at": time.Now()}) + // A heartbeat may renew only an active lease. In particular, do not + // resurrect a lease that the reconciler has already released or that has + // passed its expiry deadline. + now := time.Now() + result := common.DB.Model(&model.LiveViewerLease{}). + Where("stream_session_id = ? AND viewer_id = ? AND released_at IS NULL AND expires_at > ?", streamSessionID, userID, now). + Updates(map[string]any{"expires_at": expiresAt, "updated_at": now}) if result.Error != nil || result.RowsAffected == 0 { - return nil, common.ErrLiveNotFound + return nil, common.ErrLiveLeaseNotFound } return &vo.LiveSessionVO{Session: &session, LeaseExpiresAt: expiresAt.Unix()}, nil } @@ -204,7 +210,7 @@ func (s *LiveService) ReconcileLeases() { } for _, session := range active { if session.ExpiresAt > 0 && now.Unix() >= session.ExpiresAt { - if _, busiErr := s.stopForBilling(session.DockID, session.ID); busiErr != nil { + if _, busiErr := s.stopForBilling(session.DockID, session.ID, "session_expired"); busiErr != nil { logger.ERROR("直播会话到期停止失败: "+session.ID, busiErr) } continue @@ -223,7 +229,7 @@ func (s *LiveService) ReconcileLeases() { if session.StopDeadline.After(now) { continue } - if _, busiErr := s.stopForBilling(session.DockID, session.ID); busiErr != nil { + if _, busiErr := s.stopForBilling(session.DockID, session.ID, "no_viewers"); busiErr != nil { logger.ERROR("无观看者停止直播失败: "+session.ID, busiErr) } } @@ -361,16 +367,33 @@ func (s *LiveService) Stop(userID int64, isAdmin bool, dockID string) *common.Bu return nil } -func (s *LiveService) stopForBilling(dockID, streamSessionID string) (*model.DeviceCommandLog, *common.BusiError) { +func (s *LiveService) stopForBilling(dockID, streamSessionID string, reason string) (*model.DeviceCommandLog, *common.BusiError) { + if reason == "" { + reason = "no_balance" + } + // 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} + if reason == "no_balance" { + updates["error_code"] = "TRAFFIC_NOT_ENOUGH" + } + result := common.DB.Model(&model.LiveSession{}). + Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting')", streamSessionID, dockID). + Updates(updates) + if result.Error != nil { + return nil, common.ErrInternal + } + if result.RowsAffected == 0 { + return nil, nil + } params := map[string]any{ "streamSessionId": streamSessionID, - "reason": "no_balance", + "reason": reason, } cmd, err := DefaultCommandService.DispatchToDock(dockID, "video.stop_stream", params) if err != nil { - return nil, common.ErrInternal - } - if err := common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND phase IN ('starting','streaming','reconnecting')", streamSessionID, dockID).Updates(map[string]any{"phase": "stopping", "stop_reason": "no_balance"}).Error; err != nil { + _ = common.DB.Model(&model.LiveSession{}).Where("id = ? AND dock_id = ? AND phase = 'stopping'", streamSessionID, dockID).Updates(map[string]any{"error_code": "STOP_COMMAND_PUBLISH_FAILED"}).Error return nil, common.ErrInternal } return cmd, nil @@ -398,7 +421,7 @@ 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'", dockID) + query := common.DB.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 { @@ -440,6 +463,7 @@ func (s *LiveService) OnCommandAck(cmd *model.DeviceCommandLog, accepted bool, r } var params struct { StreamSessionID string `json:"streamSessionId"` + Reason string `json:"reason"` } if err := json.Unmarshal([]byte(cmd.Params), ¶ms); err != nil || params.StreamSessionID == "" { logger.ERROR("解析直播指令参数失败", err) @@ -453,7 +477,19 @@ func (s *LiveService) OnCommandAck(cmd *model.DeviceCommandLog, accepted bool, r if cmd.CommandType == "video.start_stream" { updates["phase"] = "failed" } else { - updates["phase"] = "streaming" + // 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" { + now := time.Now() + updates["phase"] = "stopped" + updates["device_streaming"] = false + updates["cloud_online"] = false + updates["stopped_at"] = now + } else { + updates["phase"] = "streaming" + } } if err := common.DB.Model(&model.LiveSession{}). Where("id = ? AND dock_id = ?", params.StreamSessionID, cmd.DockID). @@ -484,10 +520,10 @@ func (s *LiveService) OnVideoState(dockID string, streaming bool, streamSessionI // event that was already recorded for another live session before touching // the unique device-event column. var eventOwner model.LiveSession - if err := common.DB.Select("id").Where("device_event_id = ? AND id <> ?", eventID, streamSessionID).First(&eventOwner).Error; err == nil { - logger.WARN("忽略已被其他直播会话使用的视频状态事件", eventID, streamSessionID, eventOwner.ID) + if result := common.DB.Select("id").Where("device_event_id = ? AND id <> ?", eventID, streamSessionID).Limit(1).Find(&eventOwner); result.Error != nil { return false - } else if !errors.Is(err, gorm.ErrRecordNotFound) { + } else if result.RowsAffected > 0 { + logger.WARN("忽略已被其他直播会话使用的视频状态事件", eventID, streamSessionID, eventOwner.ID) return false } updates := map[string]any{