using System;
|
using System.Collections.Generic;
|
using System.Globalization;
|
using System.IO;
|
using System.Linq;
|
using Aging.ResourceContracts;
|
using log4net;
|
|
namespace Aging.Infrastructure.Core
|
{
|
public sealed class DbLeaseStore : ILeaseCoordinator
|
{
|
private static readonly ILog Log = LogManager.GetLogger(typeof(DbLeaseStore));
|
|
private readonly string _dbPath;
|
private readonly string _authorityId;
|
|
public string AuthorityId => _authorityId;
|
public long CurrentEpoch { get; private set; }
|
|
public DbLeaseStore(string dbPath, string authorityId)
|
{
|
_dbPath = dbPath;
|
_authorityId = authorityId;
|
EnsureDirectory();
|
InitializeSchema();
|
CurrentEpoch = ReadOrInitEpoch();
|
}
|
|
private void EnsureDirectory()
|
{
|
var dir = Path.GetDirectoryName(_dbPath);
|
if (!string.IsNullOrEmpty(dir) && !Directory.Exists(dir))
|
Directory.CreateDirectory(dir);
|
}
|
|
private void BeginImmediateTx(System.Data.SQLite.SQLiteConnection conn)
|
{
|
// BEGIN IMMEDIATE 获取 RESERVED 锁，阻止其他进程写入，防止 TOCTOU 竞态
|
using (var cmd = conn.CreateCommand())
|
{
|
cmd.CommandText = "BEGIN IMMEDIATE";
|
cmd.ExecuteNonQuery();
|
}
|
}
|
|
private void CommitTx(System.Data.SQLite.SQLiteConnection conn)
|
{
|
using (var cmd = conn.CreateCommand())
|
{
|
cmd.CommandText = "COMMIT";
|
cmd.ExecuteNonQuery();
|
}
|
}
|
|
private void RollbackTx(System.Data.SQLite.SQLiteConnection conn)
|
{
|
try
|
{
|
using (var cmd = conn.CreateCommand())
|
{
|
cmd.CommandText = "ROLLBACK";
|
cmd.ExecuteNonQuery();
|
}
|
}
|
catch { }
|
}
|
|
private System.Data.SQLite.SQLiteConnection Open()
|
{
|
var conn = new System.Data.SQLite.SQLiteConnection(
|
$"Data Source={_dbPath};Version=3;");
|
conn.Open();
|
using (var cmd = conn.CreateCommand())
|
{
|
cmd.CommandText = "PRAGMA journal_mode=WAL;";
|
cmd.ExecuteNonQuery();
|
cmd.CommandText = "PRAGMA busy_timeout=3000;";
|
cmd.ExecuteNonQuery();
|
cmd.CommandText = "PRAGMA synchronous=FULL;";
|
cmd.ExecuteNonQuery();
|
}
|
return conn;
|
}
|
|
private void InitializeSchema()
|
{
|
using (var conn = Open())
|
{
|
BeginImmediateTx(conn);
|
try
|
{
|
Execute(conn, @"
|
CREATE TABLE IF NOT EXISTS authority_meta(
|
AuthorityId TEXT PRIMARY KEY, Epoch INTEGER NOT NULL,
|
BootId TEXT NOT NULL, SchemaVersion INTEGER NOT NULL)");
|
Execute(conn, @"
|
CREATE TABLE IF NOT EXISTS resource_state(
|
ResourceKey TEXT PRIMARY KEY,
|
Generation INTEGER NOT NULL DEFAULT 0,
|
Phase INTEGER NOT NULL DEFAULT 0,
|
LeaseId TEXT, OwnerId TEXT, OwnerType INTEGER,
|
ProcessInstanceId TEXT, MachineName TEXT, HostProcessId INTEGER,
|
Epoch INTEGER NOT NULL DEFAULT 0, Reason TEXT,
|
AcquiredUtcMs INTEGER, ExpiresUtcMs INTEGER, UpdatedUtcMs INTEGER)");
|
Execute(conn, @"
|
CREATE TABLE IF NOT EXISTS lease_history(
|
EventId INTEGER PRIMARY KEY AUTOINCREMENT,
|
ResourceKey TEXT NOT NULL, Generation INTEGER NOT NULL,
|
Epoch INTEGER NOT NULL, LeaseId TEXT, OwnerId TEXT,
|
EventType TEXT NOT NULL, Reason TEXT, UtcMs INTEGER NOT NULL)");
|
Execute(conn, @"
|
CREATE TABLE IF NOT EXISTS recovery_checkpoint(
|
ResourceKey TEXT PRIMARY KEY, Epoch INTEGER NOT NULL,
|
Status INTEGER NOT NULL,
|
PhysicalReadbackJson TEXT NOT NULL, ApprovedBy TEXT,
|
EvidenceUtcMs INTEGER NOT NULL)");
|
CommitTx(conn);
|
}
|
catch
|
{
|
RollbackTx(conn);
|
throw;
|
}
|
}
|
}
|
|
private long ReadOrInitEpoch()
|
{
|
using (var conn = Open())
|
{
|
long nowMs = NowMs();
|
var row = QueryRow(conn,
|
"SELECT Epoch, BootId FROM authority_meta WHERE AuthorityId=?",
|
_authorityId);
|
if (row == null)
|
{
|
Execute(conn, "INSERT INTO authority_meta VALUES(?,?,?,1)",
|
_authorityId, 1L, Guid.NewGuid().ToString("N"));
|
return 1;
|
}
|
long oldEpoch = Convert.ToInt64(row["Epoch"]);
|
long newEpoch = oldEpoch + 1;
|
Execute(conn,
|
"UPDATE authority_meta SET Epoch=?, BootId=? WHERE AuthorityId=?",
|
newEpoch, Guid.NewGuid().ToString("N"), _authorityId);
|
Execute(conn,
|
@"INSERT INTO lease_history(ResourceKey,Generation,Epoch,LeaseId,OwnerId,EventType,Reason,UtcMs)
|
VALUES('*',0,?,'','system','EpochBump','open-authority',?)",
|
newEpoch, nowMs);
|
return newEpoch;
|
}
|
}
|
|
// ================= ILeaseCoordinator =================
|
|
public AcquireOutcome Acquire(AcquireRequest request)
|
{
|
if (request.ResourceKeys.Length == 0)
|
throw new ArgumentException("ResourceKeys 不能为空");
|
var keys = request.ResourceKeys
|
.Select(NormKey)
|
.OrderBy(k => k, StringComparer.Ordinal)
|
.Distinct()
|
.ToArray();
|
|
var deadline = NowMs() + Math.Max(0, request.WaitTimeoutMs);
|
while (true)
|
{
|
var outcome = TryAcquireOnce(keys, request);
|
if (outcome.Code != AcquireResultCode.Timeout) return outcome;
|
if (NowMs() >= deadline) { outcome.Code = AcquireResultCode.Timeout; return outcome; }
|
System.Threading.Thread.Sleep(50);
|
}
|
}
|
|
private AcquireOutcome TryAcquireOnce(string[] keys, AcquireRequest req)
|
{
|
long now = NowMs();
|
using (var conn = Open())
|
{
|
BeginImmediateTx(conn);
|
try
|
{
|
var grants = new List<LeaseGrant>(keys.Length);
|
foreach (var key in keys)
|
{
|
var st = QueryRow(conn,
|
"SELECT * FROM resource_state WHERE ResourceKey=?", key);
|
|
// RecoveryRequired → 阻止获取
|
if (st != null &&
|
Convert.ToInt32(st["Phase"]) == (int)LeasePhase.RecoveryRequired)
|
{
|
RollbackTx(conn);
|
return new AcquireOutcome
|
{
|
Code = AcquireResultCode.RecoveryBlocked,
|
RecoveryCheckpointKey = key
|
};
|
}
|
|
// Active 且未过期 → 冲突
|
bool activeHeld = st != null
|
&& Convert.ToInt32(st["Phase"]) == (int)LeasePhase.Active
|
&& Convert.ToInt64(st["ExpiresUtcMs"]) > now;
|
if (activeHeld)
|
{
|
RollbackTx(conn);
|
return new AcquireOutcome
|
{
|
Code = AcquireResultCode.ConflictHeld,
|
Holder = new ConflictInfo
|
{
|
OwnerId = st["OwnerId"]?.ToString() ?? "",
|
OwnerType = (OwnerTypeV3)Convert.ToInt32(st["OwnerType"]),
|
MachineName = st["MachineName"]?.ToString() ?? "",
|
HostProcessId = Convert.ToInt32(st["HostProcessId"]),
|
Reason = st["Reason"]?.ToString() ?? "",
|
ExpiresUtcMs = Convert.ToInt64(st["ExpiresUtcMs"])
|
}
|
};
|
}
|
|
// Active 但已过期（尚未被 Sweep）→ 阻止获取，要求先完成 Recovery
|
bool expiredNotSwept = st != null
|
&& Convert.ToInt32(st["Phase"]) == (int)LeasePhase.Active
|
&& Convert.ToInt64(st["ExpiresUtcMs"]) <= now;
|
if (expiredNotSwept)
|
{
|
RollbackTx(conn);
|
return new AcquireOutcome
|
{
|
Code = AcquireResultCode.RecoveryBlocked,
|
RecoveryCheckpointKey = key
|
};
|
}
|
|
long gen = (st != null ? Convert.ToInt64(st["Generation"]) : 0) + 1;
|
string leaseId = "lease_" + Guid.NewGuid().ToString("N").Substring(0, 12);
|
long exp = now + req.TtlSeconds * 1000L;
|
|
if (st == null)
|
{
|
Execute(conn,
|
@"INSERT INTO resource_state(ResourceKey,Generation,Phase,LeaseId,OwnerId,
|
OwnerType,ProcessInstanceId,MachineName,HostProcessId,Epoch,Reason,
|
AcquiredUtcMs,ExpiresUtcMs,UpdatedUtcMs)
|
VALUES(?,?,1,?,?,?,?,?,?,?,?,?,?,?)",
|
key, gen, leaseId, req.OwnerId, (int)req.OwnerType,
|
req.ProcessInstanceId, req.MachineName, req.HostProcessId,
|
CurrentEpoch, req.Reason, now, exp, now);
|
}
|
else
|
{
|
Execute(conn,
|
@"UPDATE resource_state SET Generation=?, Phase=1, LeaseId=?, OwnerId=?,
|
OwnerType=?, ProcessInstanceId=?, MachineName=?, HostProcessId=?,
|
Epoch=?, Reason=?, AcquiredUtcMs=?, ExpiresUtcMs=?, UpdatedUtcMs=?
|
WHERE ResourceKey=?",
|
gen, leaseId, req.OwnerId, (int)req.OwnerType, req.ProcessInstanceId,
|
req.MachineName, req.HostProcessId, CurrentEpoch, req.Reason,
|
now, exp, now, key);
|
}
|
|
Execute(conn,
|
@"INSERT INTO lease_history(ResourceKey,Generation,Epoch,LeaseId,OwnerId,EventType,Reason,UtcMs)
|
VALUES(?,?,?,?,?,'Acquire',?,?)",
|
key, gen, CurrentEpoch, leaseId, req.OwnerId, req.Reason, now);
|
|
grants.Add(new LeaseGrant(_authorityId, key, leaseId, req.OwnerId,
|
req.OwnerType, CurrentEpoch, gen, exp));
|
}
|
CommitTx(conn);
|
return new AcquireOutcome { Code = AcquireResultCode.Acquired, Grants = grants.ToArray() };
|
}
|
catch
|
{
|
RollbackTx(conn);
|
throw;
|
}
|
}
|
}
|
|
public bool Renew(LeaseGrant grant, int extendSeconds)
|
{
|
long now = NowMs();
|
using (var conn = Open())
|
{
|
BeginImmediateTx(conn);
|
try
|
{
|
int n = Execute(conn,
|
@"UPDATE resource_state SET ExpiresUtcMs=?, UpdatedUtcMs=?
|
WHERE ResourceKey=? AND LeaseId=? AND Generation=? AND Epoch=?
|
AND Phase=1 AND ExpiresUtcMs>?",
|
now + extendSeconds * 1000L, now,
|
NormKey(grant.ResourceKey), grant.LeaseId, grant.Generation, grant.Epoch, now);
|
if (n == 1)
|
CommitTx(conn);
|
else
|
{
|
RollbackTx(conn);
|
Log.Warn($"[Lease] 续期失败(栅栏不符): {grant.ResourceKey} gen={grant.Generation}");
|
}
|
return n == 1;
|
}
|
catch
|
{
|
RollbackTx(conn);
|
throw;
|
}
|
}
|
}
|
|
public bool Release(LeaseGrant grant, string note)
|
{
|
long now = NowMs();
|
using (var conn = Open())
|
{
|
BeginImmediateTx(conn);
|
try
|
{
|
int n = Execute(conn,
|
@"UPDATE resource_state SET Phase=0, LeaseId=NULL, OwnerId=NULL,
|
ExpiresUtcMs=0, UpdatedUtcMs=?
|
WHERE ResourceKey=? AND LeaseId=? AND Generation=?",
|
now, NormKey(grant.ResourceKey), grant.LeaseId, grant.Generation);
|
Execute(conn,
|
@"INSERT INTO lease_history(ResourceKey,Generation,Epoch,LeaseId,OwnerId,EventType,Reason,UtcMs)
|
SELECT ?,?,?,?,?, 'Release', ?, ? WHERE ?=1",
|
NormKey(grant.ResourceKey), grant.Generation, grant.Epoch, grant.LeaseId,
|
grant.OwnerId, note, now, n);
|
if (n == 1) CommitTx(conn); else RollbackTx(conn);
|
return n == 1;
|
}
|
catch
|
{
|
RollbackTx(conn);
|
throw;
|
}
|
}
|
}
|
|
public ConflictInfo ProbeConflict(string resourceKey)
|
{
|
long now = NowMs();
|
using (var conn = Open())
|
{
|
var st = QueryRow(conn,
|
"SELECT * FROM resource_state WHERE ResourceKey=? AND Phase=1 AND ExpiresUtcMs>?",
|
NormKey(resourceKey), now);
|
if (st == null) return null;
|
return new ConflictInfo
|
{
|
OwnerId = st["OwnerId"]?.ToString() ?? "",
|
OwnerType = (OwnerTypeV3)Convert.ToInt32(st["OwnerType"]),
|
MachineName = st["MachineName"]?.ToString() ?? "",
|
HostProcessId = Convert.ToInt32(st["HostProcessId"]),
|
Reason = st["Reason"]?.ToString() ?? "",
|
ExpiresUtcMs = Convert.ToInt64(st["ExpiresUtcMs"])
|
};
|
}
|
}
|
|
public LeasePhase PhaseOf(string resourceKey)
|
{
|
using (var conn = Open())
|
{
|
var st = QueryRow(conn,
|
"SELECT * FROM resource_state WHERE ResourceKey=?", NormKey(resourceKey));
|
if (st == null) return LeasePhase.Free;
|
int phase = Convert.ToInt32(st["Phase"]);
|
if (phase == (int)LeasePhase.RecoveryRequired) return LeasePhase.RecoveryRequired;
|
if (phase == (int)LeasePhase.Active && Convert.ToInt64(st["ExpiresUtcMs"]) <= NowMs())
|
return LeasePhase.RecoveryRequired;
|
return (LeasePhase)phase;
|
}
|
}
|
|
public int SweepExpired()
|
{
|
long now = NowMs();
|
using (var conn = Open())
|
{
|
BeginImmediateTx(conn);
|
try
|
{
|
var rows = QueryRows(conn,
|
"SELECT * FROM resource_state WHERE Phase=1 AND ExpiresUtcMs<=?", now);
|
int count = 0;
|
foreach (var st in rows)
|
{
|
string key = st["ResourceKey"]?.ToString() ?? "";
|
long gen = Convert.ToInt64(st["Generation"]);
|
long epoch = Convert.ToInt64(st["Epoch"]);
|
string leaseId = st["LeaseId"]?.ToString() ?? "";
|
string ownerId = st["OwnerId"]?.ToString() ?? "";
|
Execute(conn,
|
"UPDATE resource_state SET Phase=2, UpdatedUtcMs=? WHERE ResourceKey=?",
|
now, key);
|
Execute(conn,
|
@"INSERT INTO lease_history(ResourceKey,Generation,Epoch,LeaseId,OwnerId,EventType,Reason,UtcMs)
|
VALUES(?,?,?,?,?,'ExpiredToRecovery','ttl-missed',?)",
|
key, gen, epoch, leaseId, ownerId, now);
|
Execute(conn,
|
@"INSERT OR REPLACE INTO recovery_checkpoint(ResourceKey,Epoch,Status,PhysicalReadbackJson,ApprovedBy,EvidenceUtcMs)
|
VALUES(?,?,0,'{}',NULL,?)", key, epoch, now);
|
count++;
|
}
|
CommitTx(conn);
|
return count;
|
}
|
catch
|
{
|
RollbackTx(conn);
|
throw;
|
}
|
}
|
}
|
|
public bool ApproveRecovery(string resourceKey, string readbackJson, string approvedBy)
|
{
|
long now = NowMs();
|
using (var conn = Open())
|
{
|
BeginImmediateTx(conn);
|
try
|
{
|
Execute(conn,
|
@"INSERT OR REPLACE INTO recovery_checkpoint(ResourceKey,Epoch,Status,PhysicalReadbackJson,ApprovedBy,EvidenceUtcMs)
|
VALUES(?,?,1,?,?,?)",
|
NormKey(resourceKey), CurrentEpoch, readbackJson, approvedBy, now);
|
int n = Execute(conn,
|
@"UPDATE resource_state SET Phase=0, LeaseId=NULL, OwnerId=NULL, UpdatedUtcMs=?
|
WHERE ResourceKey=? AND Phase=2",
|
now, NormKey(resourceKey));
|
Execute(conn,
|
@"INSERT INTO lease_history(ResourceKey,Generation,Epoch,LeaseId,OwnerId,EventType,Reason,UtcMs)
|
SELECT ResourceKey,Generation,Epoch,'','system','RecoveryApproved',?,? FROM resource_state WHERE ResourceKey=?",
|
approvedBy, now, NormKey(resourceKey));
|
CommitTx(conn);
|
return n == 1;
|
}
|
catch
|
{
|
RollbackTx(conn);
|
throw;
|
}
|
}
|
}
|
|
/// <summary>当前处于待回收相位的资源键（含过期未扫的 Active）。</summary>
|
public string[] GetRecoveryRequiredKeys()
|
{
|
using (var conn = Open())
|
{
|
var rows = QueryRows(conn,
|
"SELECT ResourceKey FROM resource_state WHERE Phase=2 OR (Phase=1 AND ExpiresUtcMs<=?)",
|
NowMs());
|
return rows.Select(st => st["ResourceKey"]?.ToString() ?? "")
|
.Where(k => k.Length > 0).ToArray();
|
}
|
}
|
|
/// <summary>
|
/// 启动对账（宿主身份更替）：进程重启后旧 owner 进程已不存在，其物理 I/O 随进程自然隔离，
|
/// 按 INV-03 流程完成回收：先 SweepExpired 过期转待回收，再登记读回证据并置 Free。
|
/// 这是 MVP 单机形态的自动化形式（启动方即宿主自身）；生产多宿主应改为人工审批 UI（P2）。
|
/// </summary>
|
/// <returns>本次回收的资源数</returns>
|
public int ReconcileAfterBoot(string hostIdentity)
|
{
|
int swept = SweepExpired();
|
int approved = 0;
|
foreach (var key in GetRecoveryRequiredKeys())
|
{
|
var evidence = "{\"probe\":\"boot-reconcile\",\"host\":\"" + hostIdentity + "\"}";
|
if (ApproveRecovery(key, evidence, "boot-recovery:" + hostIdentity))
|
approved++;
|
}
|
if (swept > 0 || approved > 0)
|
Log.Info($"[DbLeaseStore] 启动对账完成: swept={swept}, approved={approved} (host={hostIdentity})");
|
return approved;
|
}
|
|
public void BumpEpochForTest(string reason)
|
{
|
using (var conn = Open())
|
{
|
BeginImmediateTx(conn);
|
try
|
{
|
// 提升 authority_meta 中的 epoch
|
Execute(conn,
|
"UPDATE authority_meta SET Epoch=Epoch+1, BootId=? WHERE AuthorityId=?",
|
Guid.NewGuid().ToString("N"), _authorityId);
|
// 在事务内读取新 epoch，用于同步更新 resource_state
|
var row = QueryRow(conn,
|
"SELECT Epoch FROM authority_meta WHERE AuthorityId=?", _authorityId);
|
long newEpoch = row != null ? Convert.ToInt64(row["Epoch"]) : CurrentEpoch;
|
// 同步提升 resource_state 中所有活跃行的 Epoch，使旧 Grant 的 Epoch 字段失效
|
Execute(conn,
|
"UPDATE resource_state SET Epoch=? WHERE Phase=1",
|
newEpoch);
|
Execute(conn,
|
@"INSERT INTO lease_history(ResourceKey,Generation,Epoch,LeaseId,OwnerId,EventType,Reason,UtcMs)
|
VALUES('*',0,?,'','system','EpochBump',?,?)",
|
newEpoch, reason, NowMs());
|
CommitTx(conn);
|
// 赋值放在 Commit 之后，保证 DB 与内存 epoch 一致
|
CurrentEpoch = newEpoch;
|
Log.Info($"[Lease] epoch 提升至 {CurrentEpoch}: {reason}");
|
}
|
catch
|
{
|
RollbackTx(conn);
|
throw;
|
}
|
}
|
}
|
|
// ================= 辅助 =================
|
|
public static string NormKey(string key) =>
|
key.Trim().ToLowerInvariant().Replace(' ', '-');
|
|
private static long NowMs() =>
|
DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
|
|
private static int Execute(System.Data.SQLite.SQLiteConnection conn, string sql,
|
params object[] args)
|
{
|
using (var cmd = conn.CreateCommand())
|
{
|
cmd.CommandText = sql;
|
foreach (var arg in args)
|
cmd.Parameters.Add(new System.Data.SQLite.SQLiteParameter { Value = arg });
|
return cmd.ExecuteNonQuery();
|
}
|
}
|
|
private static Dictionary<string, object> QueryRow(
|
System.Data.SQLite.SQLiteConnection conn, string sql, params object[] args)
|
{
|
using (var cmd = conn.CreateCommand())
|
{
|
cmd.CommandText = sql;
|
foreach (var arg in args)
|
cmd.Parameters.Add(new System.Data.SQLite.SQLiteParameter { Value = arg });
|
using (var reader = cmd.ExecuteReader())
|
{
|
if (!reader.Read()) return null;
|
var row = new Dictionary<string, object>(StringComparer.OrdinalIgnoreCase);
|
for (int i = 0; i < reader.FieldCount; i++)
|
row[reader.GetName(i)] = reader.IsDBNull(i) ? null : reader.GetValue(i);
|
return row;
|
}
|
}
|
}
|
|
private static List<Dictionary<string, object>> QueryRows(
|
System.Data.SQLite.SQLiteConnection conn, string sql, params object[] args)
|
{
|
using (var cmd = conn.CreateCommand())
|
{
|
cmd.CommandText = sql;
|
foreach (var arg in args)
|
cmd.Parameters.Add(new System.Data.SQLite.SQLiteParameter { Value = arg });
|
using (var reader = cmd.ExecuteReader())
|
{
|
var rows = new List<Dictionary<string, object>>();
|
while (reader.Read())
|
{
|
var row = new Dictionary<string, object>(StringComparer.OrdinalIgnoreCase);
|
for (int i = 0; i < reader.FieldCount; i++)
|
row[reader.GetName(i)] = reader.IsDBNull(i) ? null : reader.GetValue(i);
|
rows.Add(row);
|
}
|
return rows;
|
}
|
}
|
}
|
}
|
}
|
