zengzy / hhrcs_service · Files

using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Rcs.Application.Services;
using Rcs.Application.Services.PathFind.Models;
using Rcs.Application.Services.PathFind.Realtime;
using Rcs.Application.Services.Protocol;
using Rcs.Domain.Entities;
using Rcs.Domain.Settings;
using StackExchange.Redis;

namespace Rcs.Infrastructure.PathFinding.Realtime;

/// <summary>
/// 用于等待/排队恢复尝试的延迟调度器。
/// </summary>
public sealed class GlobalNavigationResumeScheduler : BackgroundService, IGlobalNavigationResumeScheduler
{
    /// <summary>
    /// 等待模式编码。
    /// </summary>
    private const string ModeHold = RouteModeCodes.HoldingAtWaitNode;

    /// <summary>
    /// 排队模式编码。
    /// </summary>
    private const string ModeQueue = RouteModeCodes.QueueingForJunction;

    /// <summary>
    /// 占用路口模式编码。
    /// </summary>
    private const string ModeOccupying = RouteModeCodes.OccupyingJunction;

    /// <summary>
    /// 重规划待执行模式编码。
    /// </summary>
    private const string ModeReplanPending = RouteModeCodes.ReplanPending;

    /// <summary>
    /// 阻塞模式编码。
    /// </summary>
    private const string ModeBlocked = RouteModeCodes.Blocked;

    /// <summary>
    /// 调度器日志器。
    /// </summary>
    private readonly ILogger<GlobalNavigationResumeScheduler> _logger;

    /// <summary>
    /// 用于创建临时作用域解析协议服务。
    /// </summary>
    private readonly IServiceScopeFactory _scopeFactory;

    /// <summary>
    /// Redis 连接。
    /// </summary>
    private readonly IConnectionMultiplexer _redis;

    /// <summary>
    /// 机器人状态缓存服务。
    /// </summary>
    private readonly IRobotCacheService _robotCacheService;

    /// <summary>
    /// 动态配置订阅句柄。
    /// </summary>
    private readonly IDisposable? _settingsChangeSubscription;

    /// <summary>
    /// 待恢复请求集合，Key 为请求唯一键。
    /// </summary>
    private readonly ConcurrentDictionary<string, HoldResumeRequest> _pending = new(StringComparer.OrdinalIgnoreCase);

    /// <summary>
    /// 最近一次触发调度时间，用于冷却控制。
    /// </summary>
    private readonly ConcurrentDictionary<string, DateTime> _lastDispatchUtc = new(StringComparer.OrdinalIgnoreCase);

    /// <summary>
    /// 协调器配置快照。
    /// </summary>
    private GlobalCoordinatorSettings _coordinatorSettings = new();

    /// <summary>
    /// 后台扫描间隔。
    /// </summary>
    private TimeSpan _scanInterval = TimeSpan.FromMilliseconds(300);

    /// <summary>
    /// 指标日志间隔。
    /// </summary>
    private TimeSpan _resumeMetricsInterval = TimeSpan.FromSeconds(60);

    /// <summary>
    /// 最近一次指标输出时间。
    /// </summary>
    private DateTime _lastResumeMetricsLogUtc = DateTime.Now;

    /// <summary>
    /// 已调度请求次数。
    /// </summary>
    private long _resumeScheduledCount;

    /// <summary>
    /// 成功触发恢复次数。
    /// </summary>
    private long _resumeTriggeredCount;

    /// <summary>
    /// 调度执行失败次数。
    /// </summary>
    private long _resumeDispatchFailedCount;

    /// <summary>
    /// 因冷却时间重入的次数。
    /// </summary>
    private long _resumeRequeueCooldownCount;

    /// <summary>
    /// 因等待窗口未到重入的次数。
    /// </summary>
    private long _resumeRequeueHoldWindowCount;

    /// <summary>
    /// 因机器人仍在行驶重入的次数。
    /// </summary>
    private long _resumeRequeueDrivingCount;

    /// <summary>
    /// 因等待节点不匹配重入的次数。
    /// </summary>
    private long _resumeRequeueNodeMismatchCount;

    /// <summary>
    /// 缓存缺失跳过次数。
    /// </summary>
    private long _resumeSkipCacheMissingCount;

    /// <summary>
    /// 缓存反序列化失败跳过次数。
    /// </summary>
    private long _resumeSkipDeserializeCount;

    /// <summary>
    /// 版本不匹配跳过次数。
    /// </summary>
    private long _resumeSkipVersionMismatchCount;

    /// <summary>
    /// 模式不匹配跳过次数。
    /// </summary>
    private long _resumeSkipModeMismatchCount;

    /// <summary>
    /// 调度过程异常次数。
    /// </summary>
    private long _resumeExceptionCount;

    /// <summary>
    /// 初始化恢复调度器。
    /// </summary>
    public GlobalNavigationResumeScheduler(
        ILogger<GlobalNavigationResumeScheduler> logger,
        IServiceScopeFactory scopeFactory,
        IConnectionMultiplexer redis,
        IRobotCacheService robotCacheService,
        IOptionsMonitor<AppSettings> settings)
    {
        _logger = logger;
        _scopeFactory = scopeFactory;
        _redis = redis;
        _robotCacheService = robotCacheService;
        ApplySettings(settings.CurrentValue);
        _settingsChangeSubscription = settings.OnChange(ApplySettings);
    }

    /// <summary>
    /// 提交恢复调度请求。
    /// </summary>
    /// <param name="request">恢复请求。</param>
    /// <param name="ct">取消令牌。</param>
    public Task ScheduleResumeAsync(HoldResumeRequest request, CancellationToken ct = default)
    {
        if (request.RobotId == Guid.Empty ||
            string.IsNullOrWhiteSpace(request.RobotManufacturer) ||
            string.IsNullOrWhiteSpace(request.RobotSerialNumber) ||
            string.IsNullOrWhiteSpace(request.PathCacheKey))
        {
            return Task.CompletedTask;
        }

        var normalized = CloneRequest(request);
        if (normalized.ResumeAtUtc < DateTime.Now)
        {
            normalized.ResumeAtUtc = DateTime.Now;
        }

        var key = BuildRequestKey(normalized);
        _pending.AddOrUpdate(key, normalized, (_, old) => PickPreferred(old, normalized));
        Interlocked.Increment(ref _resumeScheduledCount);
        MaybeLogResumeMetrics();
        return Task.CompletedTask;
    }

    /// <summary>
    /// 后台轮询并触发到期请求。
    /// </summary>
    /// <param name="stoppingToken">停止令牌。</param>
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            var now = DateTime.Now;
            var due = _pending
                .Where(pair => pair.Value.ResumeAtUtc <= now)
                .Select(pair => pair.Key)
                .ToList();

            foreach (var key in due)
            {
                if (!_pending.TryRemove(key, out var request))
                {
                    continue;
                }

                await DispatchIfEligibleAsync(key, request, stoppingToken);
            }

            MaybeLogResumeMetrics();
            await Task.Delay(_scanInterval, stoppingToken);
        }
    }

    /// <summary>
    /// 释放配置订阅资源。
    /// </summary>
    public override void Dispose()
    {
        _settingsChangeSubscription?.Dispose();
        base.Dispose();
    }

    /// <summary>
    /// 应用并刷新动态配置。
    /// </summary>
    /// <param name="appSettings">最新系统配置。</param>
    private void ApplySettings(AppSettings appSettings)
    {
        _coordinatorSettings = appSettings.GlobalNavigation?.Coordinator ?? new GlobalCoordinatorSettings();
        _scanInterval = TimeSpan.FromMilliseconds(Math.Clamp(_coordinatorSettings.ResumeScanIntervalMilliseconds, 100, 2000));
        _resumeMetricsInterval = TimeSpan.FromSeconds(Math.Max(10, _coordinatorSettings.DecisionMetricsIntervalSeconds));

        _logger.LogInformation(
            "[全局导航] 恢复调度器配置已更新: 扫描间隔毫秒={ScanIntervalMs}, 调度冷却秒数={DispatchCooldownSeconds}, 指标间隔秒数={MetricsIntervalSeconds}",
            _coordinatorSettings.ResumeScanIntervalMilliseconds,
            _coordinatorSettings.ResumeDispatchCooldownSeconds,
            _coordinatorSettings.DecisionMetricsIntervalSeconds);
    }

    /// <summary>
    /// 检查请求是否满足触发条件并尝试发送下一段。
    /// </summary>
    /// <param name="key">请求唯一键。</param>
    /// <param name="request">恢复请求。</param>
    /// <param name="ct">取消令牌。</param>
    private async Task DispatchIfEligibleAsync(string key, HoldResumeRequest request, CancellationToken ct)
    {
        try
        {
            var now = DateTime.Now;
            var cooldown = TimeSpan.FromSeconds(Math.Max(1, _coordinatorSettings.ResumeDispatchCooldownSeconds));
            if (_lastDispatchUtc.TryGetValue(key, out var lastDispatchUtc) &&
                now - lastDispatchUtc < cooldown)
            {
                request.ResumeAtUtc = lastDispatchUtc + cooldown;
                _pending[key] = request;
                Interlocked.Increment(ref _resumeRequeueCooldownCount);
                return;
            }

            var db = _redis.GetDatabase();
            var cacheData = await db.StringGetAsync(request.PathCacheKey);
            if (cacheData.IsNullOrEmpty)
            {
                Interlocked.Increment(ref _resumeSkipCacheMissingCount);
                return;
            }

            var cache = JsonSerializer.Deserialize<VdaSegmentedPathCache>(cacheData.ToString());
            if (cache == null)
            {
                Interlocked.Increment(ref _resumeSkipDeserializeCount);
                return;
            }

            if (cache.PlanVersion != request.ExpectedPlanVersion)
            {
                Interlocked.Increment(ref _resumeSkipVersionMismatchCount);
                return;
            }

            if (!string.Equals(cache.RouteMode, ModeHold, StringComparison.Ordinal) &&
                !string.Equals(cache.RouteMode, ModeQueue, StringComparison.Ordinal) &&
                !string.Equals(cache.RouteMode, ModeOccupying, StringComparison.Ordinal) &&
                !string.Equals(cache.RouteMode, ModeReplanPending, StringComparison.Ordinal) &&
                !string.Equals(cache.RouteMode, ModeBlocked, StringComparison.Ordinal))
            {
                Interlocked.Increment(ref _resumeSkipModeMismatchCount);
                return;
            }

            if (cache.HoldUntilUtc.HasValue && cache.HoldUntilUtc.Value > now)
            {
                request.ResumeAtUtc = cache.HoldUntilUtc.Value;
                _pending[key] = request;
                Interlocked.Increment(ref _resumeRequeueHoldWindowCount);
                return;
            }

            var status = await _robotCacheService.GetStatusAsync(request.RobotManufacturer, request.RobotSerialNumber);
            if (status?.Driving == true)
            {
                request.ResumeAtUtc = DateTime.Now.AddSeconds(1);
                _pending[key] = request;
                Interlocked.Increment(ref _resumeRequeueDrivingCount);
                return;
            }

            if (!string.IsNullOrWhiteSpace(request.HoldNodeCode) &&
                (string.IsNullOrWhiteSpace(status?.LastNode) ||
                 !string.Equals(status.LastNode, request.HoldNodeCode, StringComparison.OrdinalIgnoreCase)))
            {
                request.ResumeAtUtc = DateTime.Now.AddSeconds(1);
                _pending[key] = request;
                Interlocked.Increment(ref _resumeRequeueNodeMismatchCount);
                return;
            }

            using var scope = _scopeFactory.CreateScope();
            var protocolFactory = scope.ServiceProvider.GetRequiredService<IProtocolServiceFactory>();
            var protocol = protocolFactory.GetService(ProtocolType.VDA);
            var sendResult = await protocol.SendNextSegmentAsync(
                request.RobotManufacturer,
                request.RobotSerialNumber,
                ct: ct);

            _lastDispatchUtc[key] = DateTime.Now;

            if (!sendResult.Success)
            {
                Interlocked.Increment(ref _resumeDispatchFailedCount);
                _logger.LogDebug(
                    "[全局导航] 恢复调度失败: RobotId={RobotId}, TaskId={TaskId}, SubTaskId={SubTaskId}, Message={Message}",
                    request.RobotId, request.TaskId, request.SubTaskId, sendResult.Message);
                return;
            }

            Interlocked.Increment(ref _resumeTriggeredCount);
            _logger.LogInformation(
                "[全局导航] 已触发恢复调度: RobotId={RobotId}, TaskId={TaskId}, SubTaskId={SubTaskId}, PlanVersion={PlanVersion}",
                request.RobotId, request.TaskId, request.SubTaskId, request.ExpectedPlanVersion);
        }
        catch (OperationCanceledException)
        {
            throw;
        }
        catch (Exception ex)
        {
            Interlocked.Increment(ref _resumeExceptionCount);
            _logger.LogWarning(
                ex,
                "[全局导航] 恢复调度异常: RobotId={RobotId}, TaskId={TaskId}, SubTaskId={SubTaskId}",
                request.RobotId, request.TaskId, request.SubTaskId);
        }
    }

    /// <summary>
    /// 按配置间隔输出恢复调度指标。
    /// </summary>
    private void MaybeLogResumeMetrics()
    {
        var now = DateTime.Now;
        if (now - _lastResumeMetricsLogUtc < _resumeMetricsInterval)
        {
            return;
        }

        _lastResumeMetricsLogUtc = now;
        _logger.LogInformation(
            "[全局导航] 恢复调度指标: 已计划={Scheduled}, 已触发={Triggered}, 调度失败={DispatchFailed}, 冷却重入={RequeueCooldown}, 等待窗口重入={RequeueHoldWindow}, 行驶中重入={RequeueDriving}, 节点不匹配重入={RequeueNodeMismatch}, 跳过-缓存缺失={SkipCacheMissing}, 跳过-反序列化失败={SkipDeserialize}, 跳过-版本不匹配={SkipVersionMismatch}, 跳过-模式不匹配={SkipModeMismatch}, 异常={Exceptions}, 待处理={Pending}",
            Interlocked.Read(ref _resumeScheduledCount),
            Interlocked.Read(ref _resumeTriggeredCount),
            Interlocked.Read(ref _resumeDispatchFailedCount),
            Interlocked.Read(ref _resumeRequeueCooldownCount),
            Interlocked.Read(ref _resumeRequeueHoldWindowCount),
            Interlocked.Read(ref _resumeRequeueDrivingCount),
            Interlocked.Read(ref _resumeRequeueNodeMismatchCount),
            Interlocked.Read(ref _resumeSkipCacheMissingCount),
            Interlocked.Read(ref _resumeSkipDeserializeCount),
            Interlocked.Read(ref _resumeSkipVersionMismatchCount),
            Interlocked.Read(ref _resumeSkipModeMismatchCount),
            Interlocked.Read(ref _resumeExceptionCount),
            _pending.Count);
    }

    /// <summary>
    /// 深拷贝恢复请求对象。
    /// </summary>
    private static HoldResumeRequest CloneRequest(HoldResumeRequest source)
    {
        return new HoldResumeRequest
        {
            RobotId = source.RobotId,
            RobotManufacturer = source.RobotManufacturer,
            RobotSerialNumber = source.RobotSerialNumber,
            TaskId = source.TaskId,
            SubTaskId = source.SubTaskId,
            PathCacheKey = source.PathCacheKey,
            ExpectedPlanVersion = source.ExpectedPlanVersion,
            HoldNodeCode = source.HoldNodeCode,
            ResumeAtUtc = source.ResumeAtUtc
        };
    }

    /// <summary>
    /// 选择同一请求键下的优先请求。
    /// </summary>
    private static HoldResumeRequest PickPreferred(HoldResumeRequest oldValue, HoldResumeRequest nextValue)
    {
        if (nextValue.ExpectedPlanVersion > oldValue.ExpectedPlanVersion)
        {
            return nextValue;
        }

        if (nextValue.ExpectedPlanVersion < oldValue.ExpectedPlanVersion)
        {
            return oldValue;
        }

        return nextValue.ResumeAtUtc >= oldValue.ResumeAtUtc ? nextValue : oldValue;
    }

    /// <summary>
    /// 生成恢复请求唯一键。
    /// </summary>
    private static string BuildRequestKey(HoldResumeRequest request)
    {
        return $"{request.RobotId}:{request.TaskId}:{request.SubTaskId}:{request.PathCacheKey}";
    }
}