548 lines
28 KiB
C#
548 lines
28 KiB
C#
using dy.net.model.dto;
|
|
using dy.net.model.entity;
|
|
using dy.net.storage;
|
|
using SqlSugar;
|
|
|
|
namespace dy.net.service
|
|
{
|
|
/// <summary>
|
|
/// 将本地中转文件交给 OpenList 服务端复制。状态与外部任务 ID 持久化，应用重启后可继续。
|
|
/// </summary>
|
|
public sealed class OpenListTransferService
|
|
{
|
|
private const int MaxAttempts = 6;
|
|
private static readonly TimeSpan LeaseDuration = TimeSpan.FromSeconds(30);
|
|
private static readonly TimeSpan StallTimeout = TimeSpan.FromMinutes(60);
|
|
private static readonly TimeSpan ExternalTimeout = TimeSpan.FromHours(24);
|
|
private static readonly TimeSpan[] RetryDelays =
|
|
{
|
|
TimeSpan.FromMinutes(1),
|
|
TimeSpan.FromMinutes(5),
|
|
TimeSpan.FromMinutes(15),
|
|
TimeSpan.FromHours(1),
|
|
TimeSpan.FromHours(6)
|
|
};
|
|
|
|
private readonly ISqlSugarClient _db;
|
|
private readonly OpenListSettingsService _settingsService;
|
|
private readonly OpenListClient _client;
|
|
|
|
public OpenListTransferService(
|
|
ISqlSugarClient db,
|
|
OpenListSettingsService settingsService,
|
|
OpenListClient client)
|
|
{
|
|
_db = db;
|
|
_settingsService = settingsService;
|
|
_client = client;
|
|
}
|
|
|
|
public async Task<long> TransferAsync(
|
|
string logicalTargetPath,
|
|
Stream source,
|
|
long? declaredLength,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var settings = await _settingsService.GetAsync();
|
|
ValidateSettings(settings);
|
|
logicalTargetPath = StoragePath.NormalizeRemote(logicalTargetPath);
|
|
if (logicalTargetPath == "/") throw new ArgumentException("OpenList 文件路径不能为空。", nameof(logicalTargetPath));
|
|
var password = await _settingsService.GetPasswordAsync(settings);
|
|
if (string.IsNullOrWhiteSpace(password)) throw new InvalidOperationException("OpenList 密码无效，请重新保存配置。");
|
|
var fingerprint = StorageConfigurationFingerprint.Create(settings);
|
|
var requestedLogicalTarget = logicalTargetPath;
|
|
var actualTarget = await _client.ResolveCanonicalObjectPathAsync(settings, password,
|
|
ToActualPath(settings, logicalTargetPath), false, cancellationToken);
|
|
logicalTargetPath = ToLogicalPath(settings, actualTarget);
|
|
|
|
if (declaredLength is > 0)
|
|
{
|
|
var target = await _client.TryGetObjectAsync(settings, password, actualTarget, cancellationToken);
|
|
if (target is { IsDirectory: false } && target.Size == declaredLength.Value) return target.Size;
|
|
}
|
|
|
|
var reusable = await _db.Queryable<OpenListTransferJob>().Where(x =>
|
|
x.ConfigurationFingerprint == fingerprint
|
|
&& (x.LogicalTargetPath == logicalTargetPath || x.LogicalTargetPath == requestedLogicalTarget)
|
|
&& x.Status != OpenListTransferStatus.Succeeded
|
|
&& x.Status != OpenListTransferStatus.Failed
|
|
&& x.Status != OpenListTransferStatus.Cancelled)
|
|
.OrderByDescending(x => x.CreatedAt).FirstAsync();
|
|
var job = reusable ?? CreateJob(settings, fingerprint, logicalTargetPath, actualTarget);
|
|
if (reusable != null && (job.LogicalTargetPath != logicalTargetPath || job.ActualTargetPath != actualTarget))
|
|
{
|
|
job.LogicalTargetPath = logicalTargetPath;
|
|
job.ActualTargetPath = actualTarget;
|
|
job.UpdatedAt = DateTime.Now;
|
|
}
|
|
|
|
if (!File.Exists(job.LocalSourcePath)
|
|
|| new FileInfo(job.LocalSourcePath).Length <= 0
|
|
|| declaredLength is > 0 && new FileInfo(job.LocalSourcePath).Length != declaredLength.Value)
|
|
{
|
|
await StageSourceAsync(job, source, declaredLength, cancellationToken);
|
|
}
|
|
else if (job.ExpectedLength <= 0)
|
|
{
|
|
job.ExpectedLength = new FileInfo(job.LocalSourcePath).Length;
|
|
}
|
|
|
|
if (reusable == null) await _db.Insertable(job).ExecuteCommandAsync();
|
|
else
|
|
{
|
|
job.Status = OpenListTransferStatus.Queued;
|
|
job.NextAttemptAt = null;
|
|
job.ErrorMessage = null;
|
|
job.UpdatedAt = DateTime.Now;
|
|
await _db.Updateable(job).ExecuteCommandAsync();
|
|
}
|
|
|
|
await RunUntilSettledAsync(job.Id, cancellationToken);
|
|
var completed = await _db.Queryable<OpenListTransferJob>().InSingleAsync(job.Id);
|
|
if (completed?.Status != OpenListTransferStatus.Succeeded)
|
|
throw new IOException(completed?.ErrorMessage ?? "OpenList 服务端复制未完成。");
|
|
return completed.ExpectedLength;
|
|
}
|
|
|
|
public async Task<bool> RecoverOneAsync(CancellationToken cancellationToken)
|
|
{
|
|
var now = DateTime.Now;
|
|
var job = await _db.Queryable<OpenListTransferJob>().Where(x =>
|
|
(x.Status == OpenListTransferStatus.Queued
|
|
|| x.Status == OpenListTransferStatus.WaitingForSource
|
|
|| x.Status == OpenListTransferStatus.Copying
|
|
|| x.Status == OpenListTransferStatus.Verifying
|
|
|| x.Status == OpenListTransferStatus.Promoting
|
|
|| x.Status == OpenListTransferStatus.WaitingRetry && x.NextAttemptAt <= now)
|
|
&& (x.LeaseUntil == null || x.LeaseUntil < now))
|
|
.OrderBy(x => x.CreatedAt).FirstAsync();
|
|
if (job == null) return false;
|
|
await RunUntilSettledAsync(job.Id, cancellationToken, throwOnScheduledRetry: false);
|
|
return true;
|
|
}
|
|
|
|
public static string ToActualPath(OpenListSettings settings, string logicalPath) =>
|
|
StoragePath.CombineRemote(settings.BasePath, StoragePath.NormalizeRemote(logicalPath));
|
|
|
|
public static string ToLogicalPath(OpenListSettings settings, string actualPath)
|
|
{
|
|
var actualSegments = StoragePath.NormalizeRemote(actualPath)
|
|
.Split('/', StringSplitOptions.RemoveEmptyEntries);
|
|
var baseSegments = StoragePath.NormalizeRemote(settings.BasePath)
|
|
.Split('/', StringSplitOptions.RemoveEmptyEntries);
|
|
if (actualSegments.Length < baseSegments.Length)
|
|
throw new InvalidOperationException($"OpenList 实际路径 '{actualPath}' 不在基础目录 '{settings.BasePath}' 下。");
|
|
for (var i = 0; i < baseSegments.Length; i++)
|
|
if (!string.Equals(actualSegments[i], baseSegments[i], StringComparison.OrdinalIgnoreCase))
|
|
throw new InvalidOperationException($"OpenList 实际路径 '{actualPath}' 不在基础目录 '{settings.BasePath}' 下。");
|
|
return StoragePath.NormalizeRemote(string.Join('/', actualSegments.Skip(baseSegments.Length)));
|
|
}
|
|
|
|
public static void ValidateSettings(OpenListSettings settings)
|
|
{
|
|
if (settings == null) throw new InvalidOperationException("尚未配置 OpenList。");
|
|
_ = OpenListClient.NormalizeBaseUrl(settings.Endpoint);
|
|
if (string.IsNullOrWhiteSpace(settings.UserName)) throw new InvalidOperationException("OpenList 用户名不能为空。");
|
|
if (string.IsNullOrWhiteSpace(settings.LocalStagingPath) || !Path.IsPathRooted(settings.LocalStagingPath))
|
|
throw new InvalidOperationException("本地中转目录必须是绝对路径。");
|
|
if (string.IsNullOrWhiteSpace(settings.SourcePath)) throw new InvalidOperationException("OpenList 源挂载目录不能为空。");
|
|
_ = StoragePath.NormalizeRemote(settings.SourcePath);
|
|
_ = StoragePath.NormalizeRemote(settings.BasePath);
|
|
}
|
|
|
|
private OpenListTransferJob CreateJob(
|
|
OpenListSettings settings,
|
|
string fingerprint,
|
|
string logicalTarget,
|
|
string actualTarget)
|
|
{
|
|
var id = Guid.NewGuid().ToString("N");
|
|
var fileName = GetFileName(logicalTarget);
|
|
var localDirectory = Path.Combine(Path.GetFullPath(settings.LocalStagingPath), id);
|
|
var now = DateTime.Now;
|
|
return new OpenListTransferJob
|
|
{
|
|
Id = id,
|
|
Status = OpenListTransferStatus.Queued,
|
|
ConfigurationFingerprint = fingerprint,
|
|
LogicalTargetPath = logicalTarget,
|
|
ActualTargetPath = actualTarget,
|
|
LocalSourcePath = Path.Combine(localDirectory, fileName),
|
|
OpenListSourcePath = StoragePath.CombineRemote(settings.SourcePath, id, fileName),
|
|
StagedTargetPath = StoragePath.CombineRemote(settings.BasePath, $".dysync-staging-{id}", fileName),
|
|
CreatedAt = now,
|
|
UpdatedAt = now
|
|
};
|
|
}
|
|
|
|
private async Task StageSourceAsync(
|
|
OpenListTransferJob job,
|
|
Stream source,
|
|
long? declaredLength,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var directory = Path.GetDirectoryName(job.LocalSourcePath)
|
|
?? throw new InvalidOperationException("OpenList 本地中转目录无效。");
|
|
Directory.CreateDirectory(directory);
|
|
var temporary = job.LocalSourcePath + ".writing-" + Guid.NewGuid().ToString("N");
|
|
try
|
|
{
|
|
await using (var output = new FileStream(temporary, FileMode.CreateNew, FileAccess.Write,
|
|
FileShare.None, 81920, FileOptions.Asynchronous | FileOptions.SequentialScan))
|
|
await source.CopyToAsync(output, 81920, cancellationToken);
|
|
var length = new FileInfo(temporary).Length;
|
|
if (length <= 0) throw new MediaIntegrityException("媒体来源写入 OpenList 中转文件后为空。");
|
|
if (declaredLength is > 0 && length != declaredLength.Value)
|
|
throw new MediaIntegrityException($"下载长度不一致：期望 {declaredLength}，实际 {length}。");
|
|
File.Move(temporary, job.LocalSourcePath, true);
|
|
job.ExpectedLength = length;
|
|
}
|
|
finally
|
|
{
|
|
if (File.Exists(temporary)) File.Delete(temporary);
|
|
}
|
|
}
|
|
|
|
private async Task RunUntilSettledAsync(
|
|
string jobId,
|
|
CancellationToken cancellationToken,
|
|
bool throwOnScheduledRetry = true)
|
|
{
|
|
var leaseOwner = Guid.NewGuid().ToString("N");
|
|
if (!await TryAcquireLeaseAsync(jobId, leaseOwner))
|
|
{
|
|
await WaitForOtherProcessorAsync(jobId, cancellationToken);
|
|
return;
|
|
}
|
|
|
|
try
|
|
{
|
|
while (!cancellationToken.IsCancellationRequested)
|
|
{
|
|
var job = await _db.Queryable<OpenListTransferJob>().InSingleAsync(jobId)
|
|
?? throw new InvalidOperationException("OpenList 传输任务不存在。");
|
|
if (job.Status == OpenListTransferStatus.Succeeded) return;
|
|
if (job.Status is OpenListTransferStatus.Failed or OpenListTransferStatus.Cancelled)
|
|
throw new IOException(job.ErrorMessage ?? "OpenList 传输任务失败。");
|
|
if (job.Status == OpenListTransferStatus.WaitingRetry && job.NextAttemptAt > DateTime.Now)
|
|
{
|
|
if (throwOnScheduledRetry) throw new IOException(job.ErrorMessage ?? "OpenList 传输等待重试。");
|
|
return;
|
|
}
|
|
|
|
await RenewLeaseAsync(job, leaseOwner);
|
|
try
|
|
{
|
|
var settled = await ProcessStepAsync(job, cancellationToken);
|
|
if (settled) return;
|
|
}
|
|
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { throw; }
|
|
catch (Exception ex)
|
|
{
|
|
await ScheduleRetryOrFailAsync(job, ex.GetBaseException().Message);
|
|
if (throwOnScheduledRetry) throw;
|
|
return;
|
|
}
|
|
await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken);
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
await ReleaseLeaseAsync(jobId, leaseOwner);
|
|
}
|
|
}
|
|
|
|
private async Task<bool> ProcessStepAsync(OpenListTransferJob job, CancellationToken cancellationToken)
|
|
{
|
|
var settings = await _settingsService.GetAsync();
|
|
if (!string.Equals(StorageConfigurationFingerprint.Create(settings), job.ConfigurationFingerprint, StringComparison.Ordinal))
|
|
throw new InvalidOperationException("OpenList 配置已变化，原传输任务不会写入新目标。");
|
|
var password = await _settingsService.GetPasswordAsync(settings);
|
|
var canonicalActual = await _client.ResolveCanonicalObjectPathAsync(settings, password,
|
|
job.ActualTargetPath, false, cancellationToken);
|
|
var canonicalLogical = ToLogicalPath(settings, canonicalActual);
|
|
if (!string.Equals(job.ActualTargetPath, canonicalActual, StringComparison.Ordinal)
|
|
|| !string.Equals(job.LogicalTargetPath, canonicalLogical, StringComparison.Ordinal))
|
|
{
|
|
job.ActualTargetPath = canonicalActual;
|
|
job.LogicalTargetPath = canonicalLogical;
|
|
job.UpdatedAt = DateTime.Now;
|
|
await _db.Updateable(job).ExecuteCommandAsync();
|
|
}
|
|
var target = await _client.TryGetObjectAsync(settings, password, job.ActualTargetPath, cancellationToken);
|
|
if (target is { IsDirectory: false } && target.Size == job.ExpectedLength)
|
|
{
|
|
await MarkSucceededAsync(job, settings, password, cancellationToken);
|
|
return true;
|
|
}
|
|
|
|
if (job.Status is OpenListTransferStatus.Queued or OpenListTransferStatus.WaitingForSource
|
|
or OpenListTransferStatus.WaitingRetry)
|
|
{
|
|
if (!File.Exists(job.LocalSourcePath) || new FileInfo(job.LocalSourcePath).Length != job.ExpectedLength)
|
|
throw new FileNotFoundException("OpenList 本地中转文件不存在或长度已变化。", job.LocalSourcePath);
|
|
var source = await _client.TryGetObjectAsync(settings, password, job.OpenListSourcePath, cancellationToken);
|
|
if (source is not { IsDirectory: false } || source.Size != job.ExpectedLength)
|
|
{
|
|
job.Status = OpenListTransferStatus.WaitingForSource;
|
|
job.ErrorMessage = "OpenList 尚未看到本地中转文件，请检查源挂载目录映射。";
|
|
job.UpdatedAt = DateTime.Now;
|
|
await _db.Updateable(job).ExecuteCommandAsync();
|
|
throw new IOException(job.ErrorMessage);
|
|
}
|
|
|
|
await _client.EnsureDirectoryAsync(settings, password,
|
|
StoragePath.DirectoryName(job.ActualTargetPath), cancellationToken);
|
|
await _client.EnsureDirectoryAsync(settings, password,
|
|
StoragePath.DirectoryName(job.StagedTargetPath), cancellationToken);
|
|
var staged = await _client.TryGetObjectAsync(settings, password, job.StagedTargetPath, cancellationToken);
|
|
if (staged is { IsDirectory: false } && staged.Size == job.ExpectedLength)
|
|
{
|
|
job.Status = OpenListTransferStatus.Verifying;
|
|
job.ErrorMessage = null;
|
|
job.UpdatedAt = DateTime.Now;
|
|
await _db.Updateable(job).ExecuteCommandAsync();
|
|
return false;
|
|
}
|
|
if (staged != null) await _client.DeleteObjectAsync(settings, password, job.StagedTargetPath, cancellationToken);
|
|
var copy = await _client.CopyFileAsync(settings, password,
|
|
job.OpenListSourcePath, job.StagedTargetPath, cancellationToken);
|
|
job.Attempts++;
|
|
job.ExternalTaskId = copy.TaskIds.FirstOrDefault();
|
|
job.ExternalTaskStartedAt = DateTime.Now;
|
|
job.LastProgressAt = DateTime.Now;
|
|
job.ExternalProgress = 0;
|
|
job.Status = string.IsNullOrWhiteSpace(job.ExternalTaskId)
|
|
? OpenListTransferStatus.Verifying
|
|
: OpenListTransferStatus.Copying;
|
|
job.ErrorMessage = null;
|
|
job.NextAttemptAt = null;
|
|
job.UpdatedAt = DateTime.Now;
|
|
await _db.Updateable(job).ExecuteCommandAsync();
|
|
return false;
|
|
}
|
|
|
|
if (job.Status == OpenListTransferStatus.Copying)
|
|
{
|
|
if (job.ExternalTaskStartedAt.HasValue && DateTime.Now - job.ExternalTaskStartedAt.Value > ExternalTimeout)
|
|
{
|
|
_ = await _client.TryCancelCopyTaskAsync(settings, password, job.ExternalTaskId, cancellationToken);
|
|
throw new TimeoutException("OpenList 复制任务运行超过 24 小时。");
|
|
}
|
|
if (job.LastProgressAt.HasValue && DateTime.Now - job.LastProgressAt.Value > StallTimeout)
|
|
{
|
|
var cancelled = await _client.TryCancelCopyTaskAsync(settings, password, job.ExternalTaskId, cancellationToken);
|
|
throw new TimeoutException(cancelled
|
|
? "OpenList 复制任务连续 60 分钟没有进度，已取消。"
|
|
: "OpenList 复制任务连续 60 分钟没有进度且无法取消。");
|
|
}
|
|
var task = await _client.TryGetCopyTaskAsync(settings, password, job.ExternalTaskId, cancellationToken);
|
|
if (task == null)
|
|
{
|
|
var staged = await _client.TryGetObjectAsync(settings, password, job.StagedTargetPath, cancellationToken);
|
|
if (staged is not { IsDirectory: false } || staged.Size != job.ExpectedLength)
|
|
throw new IOException($"OpenList 复制任务不存在：{job.ExternalTaskId}");
|
|
job.Status = OpenListTransferStatus.Verifying;
|
|
}
|
|
else if (task.State == 2) job.Status = OpenListTransferStatus.Verifying;
|
|
else if (task.State is 4 or 7)
|
|
throw new IOException(string.IsNullOrWhiteSpace(task.Error)
|
|
? $"OpenList 复制任务以状态 {task.State} 结束。" : task.Error);
|
|
else if (Math.Abs(task.Progress - job.ExternalProgress) > 0.001)
|
|
{
|
|
job.ExternalProgress = task.Progress;
|
|
job.LastProgressAt = DateTime.Now;
|
|
}
|
|
job.UpdatedAt = DateTime.Now;
|
|
await _db.Updateable(job).ExecuteCommandAsync();
|
|
return false;
|
|
}
|
|
|
|
if (job.Status == OpenListTransferStatus.Verifying)
|
|
{
|
|
var staged = await _client.TryGetObjectAsync(settings, password, job.StagedTargetPath, cancellationToken);
|
|
if (staged is not { IsDirectory: false } || staged.Size != job.ExpectedLength)
|
|
throw new IOException($"OpenList 暂存文件长度校验失败：期望 {job.ExpectedLength}，实际 {staged?.Size.ToString() ?? "不存在"}。");
|
|
job.Status = OpenListTransferStatus.Promoting;
|
|
job.UpdatedAt = DateTime.Now;
|
|
await _db.Updateable(job).ExecuteCommandAsync();
|
|
return false;
|
|
}
|
|
|
|
if (job.Status == OpenListTransferStatus.Promoting)
|
|
{
|
|
await PromoteAsync(job, settings, password, cancellationToken);
|
|
await MarkSucceededAsync(job, settings, password, cancellationToken);
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
private async Task PromoteAsync(
|
|
OpenListTransferJob job,
|
|
OpenListSettings settings,
|
|
string password,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var target = await _client.TryGetObjectAsync(settings, password, job.ActualTargetPath, cancellationToken);
|
|
if (target is { IsDirectory: false } && target.Size == job.ExpectedLength) return;
|
|
|
|
if (target != null && string.IsNullOrWhiteSpace(job.BackupTargetPath))
|
|
{
|
|
var backupName = $".dysync-old-{job.Id[..8]}-{GetFileName(job.ActualTargetPath)}";
|
|
await _client.RenameAsync(settings, password, job.ActualTargetPath, backupName, cancellationToken);
|
|
job.BackupTargetPath = StoragePath.CombineRemote(
|
|
StoragePath.DirectoryName(job.ActualTargetPath), backupName);
|
|
job.UpdatedAt = DateTime.Now;
|
|
await _db.Updateable(job).ExecuteCommandAsync();
|
|
}
|
|
|
|
try
|
|
{
|
|
var final = await _client.TryGetObjectAsync(settings, password, job.ActualTargetPath, cancellationToken);
|
|
if (final == null)
|
|
await _client.MoveAsync(settings, password, job.StagedTargetPath,
|
|
StoragePath.DirectoryName(job.ActualTargetPath), false, cancellationToken);
|
|
final = await _client.TryGetObjectAsync(settings, password, job.ActualTargetPath, cancellationToken);
|
|
if (final is not { IsDirectory: false } || final.Size != job.ExpectedLength)
|
|
throw new IOException("OpenList 暂存文件提升后最终目标校验失败。");
|
|
if (!string.IsNullOrWhiteSpace(job.BackupTargetPath))
|
|
{
|
|
await _client.DeleteObjectAsync(settings, password, job.BackupTargetPath, cancellationToken);
|
|
job.BackupTargetPath = null;
|
|
}
|
|
}
|
|
catch
|
|
{
|
|
var final = await _client.TryGetObjectAsync(settings, password, job.ActualTargetPath, cancellationToken);
|
|
var backup = string.IsNullOrWhiteSpace(job.BackupTargetPath) ? null
|
|
: await _client.TryGetObjectAsync(settings, password, job.BackupTargetPath, cancellationToken);
|
|
if (final == null && backup != null)
|
|
{
|
|
await _client.RenameAsync(settings, password, job.BackupTargetPath,
|
|
GetFileName(job.ActualTargetPath), cancellationToken);
|
|
job.BackupTargetPath = null;
|
|
job.UpdatedAt = DateTime.Now;
|
|
await _db.Updateable(job).ExecuteCommandAsync();
|
|
}
|
|
throw;
|
|
}
|
|
}
|
|
|
|
private async Task MarkSucceededAsync(
|
|
OpenListTransferJob job,
|
|
OpenListSettings settings,
|
|
string password,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var final = await _client.TryGetObjectAsync(settings, password, job.ActualTargetPath, cancellationToken);
|
|
if (final is not { IsDirectory: false } || final.Size != job.ExpectedLength)
|
|
throw new IOException("OpenList 最终目标不存在或长度不一致。");
|
|
job.Status = OpenListTransferStatus.Succeeded;
|
|
job.ErrorMessage = null;
|
|
job.NextAttemptAt = null;
|
|
job.ExternalTaskId = null;
|
|
job.CompletedAt = DateTime.Now;
|
|
job.UpdatedAt = DateTime.Now;
|
|
await _db.Updateable(job).ExecuteCommandAsync();
|
|
CleanupLocal(job.LocalSourcePath);
|
|
try
|
|
{
|
|
var stagedDirectory = StoragePath.DirectoryName(job.StagedTargetPath);
|
|
await _client.DeleteObjectAsync(settings, password, stagedDirectory, cancellationToken);
|
|
}
|
|
catch (Exception ex) { Serilog.Log.Debug(ex, "清理 OpenList 暂存目录失败：{Path}", job.StagedTargetPath); }
|
|
}
|
|
|
|
private async Task ScheduleRetryOrFailAsync(OpenListTransferJob job, string error)
|
|
{
|
|
job.ExternalTaskId = null;
|
|
job.ExternalProgress = 0;
|
|
job.ExternalTaskStartedAt = null;
|
|
job.LastProgressAt = null;
|
|
job.ErrorMessage = error;
|
|
job.UpdatedAt = DateTime.Now;
|
|
if (job.Attempts >= MaxAttempts)
|
|
{
|
|
job.Status = OpenListTransferStatus.Failed;
|
|
job.CompletedAt = DateTime.Now;
|
|
job.NextAttemptAt = null;
|
|
}
|
|
else
|
|
{
|
|
var index = Math.Clamp(Math.Max(1, job.Attempts) - 1, 0, RetryDelays.Length - 1);
|
|
job.Status = OpenListTransferStatus.WaitingRetry;
|
|
job.NextAttemptAt = DateTime.Now.Add(RetryDelays[index]);
|
|
}
|
|
await _db.Updateable(job).ExecuteCommandAsync();
|
|
}
|
|
|
|
private async Task<bool> TryAcquireLeaseAsync(string jobId, string owner)
|
|
{
|
|
var now = DateTime.Now;
|
|
var changed = await _db.Updateable<OpenListTransferJob>()
|
|
.SetColumns(x => new OpenListTransferJob
|
|
{
|
|
LeaseOwner = owner,
|
|
LeaseUntil = now.Add(LeaseDuration),
|
|
UpdatedAt = now
|
|
})
|
|
.Where(x => x.Id == jobId && (x.LeaseUntil == null || x.LeaseUntil < now))
|
|
.ExecuteCommandAsync();
|
|
return changed == 1;
|
|
}
|
|
|
|
private async Task RenewLeaseAsync(OpenListTransferJob job, string owner)
|
|
{
|
|
job.LeaseOwner = owner;
|
|
job.LeaseUntil = DateTime.Now.Add(LeaseDuration);
|
|
job.UpdatedAt = DateTime.Now;
|
|
await _db.Updateable(job).ExecuteCommandAsync();
|
|
}
|
|
|
|
private async Task ReleaseLeaseAsync(string jobId, string owner)
|
|
{
|
|
await _db.Updateable<OpenListTransferJob>()
|
|
.SetColumns(x => new OpenListTransferJob { LeaseOwner = null, LeaseUntil = null, UpdatedAt = DateTime.Now })
|
|
.Where(x => x.Id == jobId && x.LeaseOwner == owner).ExecuteCommandAsync();
|
|
}
|
|
|
|
private async Task WaitForOtherProcessorAsync(string jobId, CancellationToken cancellationToken)
|
|
{
|
|
while (!cancellationToken.IsCancellationRequested)
|
|
{
|
|
var job = await _db.Queryable<OpenListTransferJob>().InSingleAsync(jobId)
|
|
?? throw new InvalidOperationException("OpenList 传输任务不存在。");
|
|
if (job.Status == OpenListTransferStatus.Succeeded) return;
|
|
if (job.Status is OpenListTransferStatus.Failed or OpenListTransferStatus.Cancelled
|
|
or OpenListTransferStatus.WaitingRetry)
|
|
throw new IOException(job.ErrorMessage ?? "OpenList 传输未完成。");
|
|
if (job.LeaseUntil == null || job.LeaseUntil < DateTime.Now)
|
|
{
|
|
await RunUntilSettledAsync(jobId, cancellationToken);
|
|
return;
|
|
}
|
|
await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken);
|
|
}
|
|
}
|
|
|
|
private static void CleanupLocal(string path)
|
|
{
|
|
try
|
|
{
|
|
if (File.Exists(path)) File.Delete(path);
|
|
var directory = Path.GetDirectoryName(path);
|
|
if (!string.IsNullOrWhiteSpace(directory) && Directory.Exists(directory)
|
|
&& !Directory.EnumerateFileSystemEntries(directory).Any()) Directory.Delete(directory);
|
|
}
|
|
catch (Exception ex) { Serilog.Log.Warning(ex, "清理 OpenList 本地中转文件失败：{Path}", path); }
|
|
}
|
|
|
|
private static string GetFileName(string path)
|
|
{
|
|
var normalized = StoragePath.NormalizeRemote(path);
|
|
return normalized[(normalized.LastIndexOf('/') + 1)..];
|
|
}
|
|
}
|
|
}
|