using System.Collections.Concurrent;
|
using Aging.Infrastructure.Contracts;
|
using Aging.ResourceContracts;
|
using log4net;
|
|
namespace Aging.Infrastructure.Core
|
{
|
/// <summary>
|
/// 基于 DbLeaseStore 的 IResourceManager 分层实现。
|
/// 对外签名与 ResourceManager 完全一致（零改动），内部转发到 ILeaseCoordinator 仲裁库。
|
/// </summary>
|
public sealed class LayeredResourceManager : IResourceManager
|
{
|
private static readonly ILog Log = LogManager.GetLogger(typeof(LayeredResourceManager));
|
|
private readonly ILeaseCoordinator _lease;
|
|
/// <summary>
|
/// 按 TokenId 缓存已获取的 LeaseGrant 数组，供 ValidateOwner / Release / AcquireOperation 使用。
|
/// </summary>
|
private readonly ConcurrentDictionary<string, LeaseGrant[]> _grantsByToken = new();
|
|
/// <summary>
|
/// 默认租约 TTL（秒）
|
/// </summary>
|
private const int DefaultTtlSeconds = 300; // 5 分钟，足够长用于 MVP
|
|
public LayeredResourceManager(ILeaseCoordinator lease)
|
{
|
_lease = lease ?? throw new ArgumentNullException(nameof(lease));
|
}
|
|
/// <summary>
|
/// 获取底层的 ILeaseCoordinator（供 AutoDeviceLease 心跳续期使用）
|
/// </summary>
|
public ILeaseCoordinator Coordinator => _lease;
|
|
// ======================== ReserveAsync ========================
|
|
public async Task<ResourceReservation> 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<IOperationLease> 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);
|
}
|
|
// ======================== 内部辅助 ========================
|
|
/// <summary>
|
/// 根据 TokenId 获取缓存的 LeaseGrant 数组（供 AutoDeviceLease 心跳续期使用）
|
/// </summary>
|
public bool TryGetGrants(string tokenId, out LeaseGrant[] grants)
|
{
|
return _grantsByToken.TryGetValue(tokenId, out grants);
|
}
|
|
/// <summary>
|
/// 根据 TokenId 和资源键查找对应的 LeaseGrant
|
/// </summary>
|
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;
|
}
|
|
/// <summary>
|
/// 构造冲突异常消息
|
/// </summary>
|
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）";
|
}
|
}
|
}
|
