using System.Collections.Concurrent; using Aging.Infrastructure.Contracts; using Aging.ResourceContracts; using log4net; namespace Aging.Infrastructure.Core { /// /// 基于 DbLeaseStore 的 IResourceManager 分层实现。 /// 对外签名与 ResourceManager 完全一致(零改动),内部转发到 ILeaseCoordinator 仲裁库。 /// public sealed class LayeredResourceManager : IResourceManager { private static readonly ILog Log = LogManager.GetLogger(typeof(LayeredResourceManager)); private readonly ILeaseCoordinator _lease; /// /// 按 TokenId 缓存已获取的 LeaseGrant 数组,供 ValidateOwner / Release / AcquireOperation 使用。 /// private readonly ConcurrentDictionary _grantsByToken = new(); /// /// 默认租约 TTL(秒) /// private const int DefaultTtlSeconds = 300; // 5 分钟,足够长用于 MVP public LayeredResourceManager(ILeaseCoordinator lease) { _lease = lease ?? throw new ArgumentNullException(nameof(lease)); } /// /// 获取底层的 ILeaseCoordinator(供 AutoDeviceLease 心跳续期使用) /// public ILeaseCoordinator Coordinator => _lease; // ======================== ReserveAsync ======================== public async Task ReserveAsync( ResourceOwnerToken owner, ResourceKey resource, CancellationToken ct) { if (owner == null) throw new ArgumentNullException(nameof(owner)); if (resource == null) throw new ArgumentNullException(nameof(resource)); var req = new AcquireRequest { OwnerId = owner.OwnerId, OwnerType = (OwnerTypeV3)(int)owner.OwnerType, ResourceKeys = new[] { resource.ToString() }, TtlSeconds = DefaultTtlSeconds, WaitTimeoutMs = 0, Reason = owner.OwnerId }; AcquireOutcome outcome; try { outcome = await Task.Run(() => _lease.Acquire(req), ct).ConfigureAwait(false); } catch (OperationCanceledException) { throw; } catch (Exception ex) { Log.Error($"[LayeredRM] Acquire 异常: {ex.Message}", ex); throw new ResourceConflictException($"资源 {resource} 获取异常: {ex.Message}", ex); } switch (outcome.Code) { case AcquireResultCode.Acquired: case AcquireResultCode.Reused: _grantsByToken[owner.TokenId] = outcome.Grants; Log.Info($"[LayeredRM] 资源 {resource} 预约成功,所有者: {owner.OwnerId},lease: {outcome.Grants[0].LeaseId}"); return new ResourceReservation(resource, owner) { State = ReservationState.Active, Description = $"LeaseId={outcome.Grants[0].LeaseId}" }; case AcquireResultCode.ConflictHeld: case AcquireResultCode.Timeout: var msg = BuildConflictMessage(resource, outcome.Holder); Log.Warn($"[LayeredRM] 资源冲突: {msg}"); throw new ResourceConflictException(msg); case AcquireResultCode.RecoveryBlocked: throw new ResourceConflictException( $"资源 {resource} 处于恢复中(RecoveryRequired),暂不可用"); default: throw new ResourceConflictException($"资源 {resource} 获取失败: {outcome.Code}"); } } // ======================== ReleaseReservationAsync ======================== public async Task ReleaseReservationAsync( ResourceOwnerToken owner, ResourceKey resource, CancellationToken ct) { if (owner == null) throw new ArgumentNullException(nameof(owner)); if (resource == null) throw new ArgumentNullException(nameof(resource)); var grant = FindGrant(owner.TokenId, resource); if (grant == null) { Log.Warn($"[LayeredRM] 尝试释放无 grant 的资源: {resource}, owner: {owner.OwnerId}"); return; } await Task.Run(() => _lease.Release(grant, owner.OwnerId), ct).ConfigureAwait(false); // 清除缓存 _grantsByToken.TryRemove(owner.TokenId, out _); Log.Info($"[LayeredRM] 资源 {resource} 已释放,所有者: {owner.OwnerId}"); } // ======================== ValidateOwner ======================== public bool ValidateOwner(ResourceOwnerToken owner, ResourceKey resource) { if (owner == null) throw new ArgumentNullException(nameof(owner)); if (resource == null) throw new ArgumentNullException(nameof(resource)); var grant = FindGrant(owner.TokenId, resource); if (grant == null) return false; // 栅栏校验:Renew 仅在“租约仍 Active 且 generation/epoch 匹配且未过期”时返回 true。 // 用完整 TTL 续期而非 1s:既校验归属,又作为每次操作的保活, // 避免把租约到期时间砍到 1s 导致后续心跳 Renew 因“已过期”而失败→误报租约失效。 return _lease.Renew(grant, DefaultTtlSeconds); } // ======================== IsReserved ======================== public bool IsReserved(ResourceKey resource, out ResourceReservation? reservation) { if (resource == null) throw new ArgumentNullException(nameof(resource)); var conflict = _lease.ProbeConflict(resource.ToString()); if (conflict != null) { // 构造一个 "占位" reservation 返回给调用方 var syntheticOwner = new ResourceOwnerToken( (OwnerType)(int)conflict.OwnerType, conflict.OwnerId); reservation = new ResourceReservation(resource, syntheticOwner, $"占用中: {conflict.OwnerId} ({conflict.Reason})") { State = ReservationState.Active }; return true; } reservation = null; return false; } // ======================== AcquireOperationAsync ======================== public async Task AcquireOperationAsync( ResourceOwnerToken owner, ResourceKey resource, CancellationToken ct) { if (owner == null) throw new ArgumentNullException(nameof(owner)); if (resource == null) throw new ArgumentNullException(nameof(resource)); // 栅栏校验:确保 owner 仍持有该资源 if (!ValidateOwner(owner, resource)) { throw new InvalidOperationException( $"资源 {resource} 的所有者栅栏校验失败,owner: {owner.OwnerId}"); } await Task.CompletedTask; return new OperationLease( resource, owner, onRelease: () => Log.Debug($"[LayeredRM] 操作租约释放: {resource}"), leaseDurationSeconds: 30); } // ======================== 内部辅助 ======================== /// /// 根据 TokenId 获取缓存的 LeaseGrant 数组(供 AutoDeviceLease 心跳续期使用) /// public bool TryGetGrants(string tokenId, out LeaseGrant[] grants) { return _grantsByToken.TryGetValue(tokenId, out grants); } /// /// 根据 TokenId 和资源键查找对应的 LeaseGrant /// private LeaseGrant? FindGrant(string tokenId, ResourceKey resource) { if (!_grantsByToken.TryGetValue(tokenId, out var grants)) return null; var resourceStr = resource.ToString(); foreach (var g in grants) { // DbLeaseStore 对 key 做了 NormKey(trim+lower+空格替换为-), // 这里用相同的规范化来比较 if (string.Equals(DbLeaseStore.NormKey(g.ResourceKey), DbLeaseStore.NormKey(resourceStr), StringComparison.OrdinalIgnoreCase)) return g; } return null; } /// /// 构造冲突异常消息 /// private static string BuildConflictMessage(ResourceKey resource, ConflictInfo? holder) { if (holder == null) return $"资源 {resource} 正被占用"; long nowMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); long remaining = Math.Max(0, (holder.ExpiresUtcMs - nowMs) / 1000); return $"资源 {resource} 正被占用:{holder.OwnerId}" + $"({holder.OwnerType},机器 {holder.MachineName} PID {holder.HostProcessId}," + $"剩余租约 {remaining}s)"; } } }