484 lines
18 KiB
C#
484 lines
18 KiB
C#
using LightlessSyncShared.Metrics;
|
|
using LightlessSyncShared.Services;
|
|
using LightlessSyncStaticFilesServer.Utils;
|
|
using System.Collections.Concurrent;
|
|
using System.Net.Http.Headers;
|
|
using LightlessSyncShared.Utils;
|
|
using LightlessSync.API.Routes;
|
|
using LightlessSyncShared.Utils.Configuration;
|
|
|
|
namespace LightlessSyncStaticFilesServer.Services;
|
|
|
|
public sealed class CachedFileProvider : IDisposable
|
|
{
|
|
private readonly IConfigurationService<StaticFilesServerConfiguration> _configuration;
|
|
private readonly ILogger<CachedFileProvider> _logger;
|
|
private readonly FileStatisticsService _fileStatisticsService;
|
|
private readonly LightlessMetrics _metrics;
|
|
private readonly ServerTokenGenerator _generator;
|
|
private readonly IShardFileInventoryReporter _inventoryReporter;
|
|
private readonly Uri _remoteCacheSourceUri;
|
|
private readonly string _hotStoragePath;
|
|
private readonly ConcurrentDictionary<string, Task> _currentTransfers = new(StringComparer.Ordinal);
|
|
private readonly HttpClient _httpClient;
|
|
private readonly SemaphoreSlim _downloadSemaphore = new(1, 1);
|
|
private bool _disposed;
|
|
|
|
private bool IsMainServer => _remoteCacheSourceUri == null && _isDistributionServer;
|
|
private bool _isDistributionServer;
|
|
|
|
public CachedFileProvider(IConfigurationService<StaticFilesServerConfiguration> configuration, ILogger<CachedFileProvider> logger,
|
|
FileStatisticsService fileStatisticsService, LightlessMetrics metrics, ServerTokenGenerator generator,
|
|
IShardFileInventoryReporter inventoryReporter)
|
|
{
|
|
_configuration = configuration;
|
|
_logger = logger;
|
|
_fileStatisticsService = fileStatisticsService;
|
|
_metrics = metrics;
|
|
_generator = generator;
|
|
_inventoryReporter = inventoryReporter;
|
|
_remoteCacheSourceUri = configuration.GetValueOrDefault<Uri>(nameof(StaticFilesServerConfiguration.DistributionFileServerAddress), null);
|
|
_isDistributionServer = configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.IsDistributionNode), false);
|
|
_hotStoragePath = configuration.GetValue<string>(nameof(StaticFilesServerConfiguration.CacheDirectory));
|
|
_httpClient = new();
|
|
_httpClient.DefaultRequestHeaders.UserAgent.Add(new ProductInfoHeaderValue("LightlessSyncServer", "1.0.0.0"));
|
|
_httpClient.Timeout = TimeSpan.FromSeconds(300);
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
if (_disposed)
|
|
{
|
|
return;
|
|
}
|
|
|
|
_disposed = true;
|
|
_httpClient?.Dispose();
|
|
}
|
|
|
|
private async Task DownloadTask(string hash)
|
|
{
|
|
var destinationFilePath = FilePathUtil.GetFilePath(_hotStoragePath, hash);
|
|
|
|
// first check cold storage
|
|
if (TryCopyFromColdStorage(hash, destinationFilePath)) return;
|
|
|
|
// if cold storage is not configured or file not found or error is present try to download file from remote
|
|
var downloadUrl = LightlessFiles.DistributionGetFullPath(_remoteCacheSourceUri, hash);
|
|
_logger.LogInformation("Did not find {hash}, downloading from {server}", hash, downloadUrl);
|
|
|
|
using var requestMessage = new HttpRequestMessage(HttpMethod.Get, downloadUrl);
|
|
requestMessage.Headers.Authorization = new AuthenticationHeaderValue("Bearer", _generator.Token);
|
|
HttpResponseMessage? response = null;
|
|
|
|
try
|
|
{
|
|
response = await _httpClient.SendAsync(requestMessage).ConfigureAwait(false);
|
|
response.EnsureSuccessStatusCode();
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Failed to download {url}", downloadUrl);
|
|
response?.Dispose();
|
|
return;
|
|
}
|
|
|
|
var tempFileName = destinationFilePath + ".dl";
|
|
var fileStream = new FileStream(tempFileName, FileMode.Create, FileAccess.ReadWrite);
|
|
var bufferSize = response.Content.Headers.ContentLength > 1024 * 1024 ? 4096 : 1024;
|
|
var buffer = new byte[bufferSize];
|
|
|
|
var bytesRead = 0;
|
|
using var content = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
|
|
while ((bytesRead = await content.ReadAsync(buffer).ConfigureAwait(false)) > 0)
|
|
{
|
|
await fileStream.WriteAsync(buffer.AsMemory(0, bytesRead)).ConfigureAwait(false);
|
|
}
|
|
await fileStream.FlushAsync().ConfigureAwait(false);
|
|
await fileStream.DisposeAsync().ConfigureAwait(false);
|
|
File.Move(tempFileName, destinationFilePath, true);
|
|
|
|
_metrics.IncGauge(MetricsAPI.GaugeFilesTotal);
|
|
_metrics.IncGauge(MetricsAPI.GaugeFilesTotalSize, FilePathUtil.GetFileInfoForHash(_hotStoragePath, hash).Length);
|
|
_inventoryReporter.ReportAdded(hash);
|
|
response.Dispose();
|
|
}
|
|
|
|
private bool TryCopyFromColdStorage(string hash, string destinationFilePath)
|
|
{
|
|
if (!_configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.UseColdStorage), false)) return false;
|
|
|
|
string coldStorageDir = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.ColdStorageDirectory), string.Empty);
|
|
if (string.IsNullOrEmpty(coldStorageDir)) return false;
|
|
|
|
var coldStorageFilePath = FilePathUtil.GetFileInfoForHash(coldStorageDir, hash);
|
|
if (coldStorageFilePath == null) return false;
|
|
|
|
try
|
|
{
|
|
_logger.LogDebug("Copying {hash} from cold storage: {path}", hash, coldStorageFilePath);
|
|
var tempFileName = destinationFilePath + ".dl";
|
|
File.Copy(coldStorageFilePath.FullName, tempFileName, true);
|
|
File.Move(tempFileName, destinationFilePath, true);
|
|
|
|
File.SetLastAccessTimeUtc(coldStorageFilePath.FullName, DateTime.UtcNow);
|
|
File.SetLastAccessTimeUtc(destinationFilePath, DateTime.UtcNow);
|
|
File.SetCreationTimeUtc(destinationFilePath, DateTime.UtcNow);
|
|
File.SetLastWriteTimeUtc(destinationFilePath, DateTime.UtcNow);
|
|
|
|
_metrics.IncGauge(MetricsAPI.GaugeFilesTotal);
|
|
_metrics.IncGauge(MetricsAPI.GaugeFilesTotalSize, new FileInfo(destinationFilePath).Length);
|
|
_inventoryReporter.ReportAdded(hash);
|
|
return true;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Could not copy {coldStoragePath} from cold storage", coldStorageFilePath);
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
public async Task<CachedFileStreamResult?> GetFileStreamForDirectDownloadAsync(string hash, CancellationToken ct)
|
|
{
|
|
var destinationFilePath = FilePathUtil.GetFilePath(_hotStoragePath, hash);
|
|
|
|
var existing = GetLocalFilePath(hash);
|
|
if (existing != null)
|
|
{
|
|
return new CachedFileStreamResult(existing, null, existing.Length);
|
|
}
|
|
|
|
if (TryCopyFromColdStorage(hash, destinationFilePath))
|
|
{
|
|
var coldFile = GetLocalFilePath(hash);
|
|
if (coldFile != null)
|
|
{
|
|
return new CachedFileStreamResult(coldFile, null, coldFile.Length);
|
|
}
|
|
}
|
|
|
|
if (_remoteCacheSourceUri == null)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
TaskCompletionSource<bool>? completion = null;
|
|
|
|
await _downloadSemaphore.WaitAsync(ct).ConfigureAwait(false);
|
|
try
|
|
{
|
|
if (_currentTransfers.TryGetValue(hash, out var downloadTask)
|
|
&& !(downloadTask?.IsCompleted ?? true))
|
|
{
|
|
completion = null;
|
|
}
|
|
else
|
|
{
|
|
completion = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
_currentTransfers[hash] = completion.Task;
|
|
_metrics.IncGauge(MetricsAPI.GaugeFilesDownloadingFromCache);
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
_downloadSemaphore.Release();
|
|
}
|
|
|
|
if (completion == null)
|
|
{
|
|
var waited = await DownloadAndGetLocalFileInfo(hash).ConfigureAwait(false);
|
|
if (waited == null) return null;
|
|
return new CachedFileStreamResult(waited, null, waited.Length);
|
|
}
|
|
|
|
var downloadUrl = LightlessFiles.DistributionGetFullPath(_remoteCacheSourceUri, hash);
|
|
_logger.LogInformation("Did not find {hash}, streaming from {server}", hash, downloadUrl);
|
|
|
|
using var requestMessage = new HttpRequestMessage(HttpMethod.Get, downloadUrl);
|
|
requestMessage.Headers.Authorization = new AuthenticationHeaderValue("Bearer", _generator.Token);
|
|
|
|
HttpResponseMessage response;
|
|
try
|
|
{
|
|
response = await _httpClient.SendAsync(requestMessage, HttpCompletionOption.ResponseHeadersRead, ct).ConfigureAwait(false);
|
|
response.EnsureSuccessStatusCode();
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Failed to stream {url}", downloadUrl);
|
|
FinalizeStreamingDownload(hash, null, destinationFilePath, completion, false, 0);
|
|
return null;
|
|
}
|
|
|
|
var tempFileName = destinationFilePath + ".dl";
|
|
var fileStream = new FileStream(tempFileName, FileMode.Create, FileAccess.Write, FileShare.Read, bufferSize: 64 * 1024, useAsync: true);
|
|
var sourceStream = await response.Content.ReadAsStreamAsync(ct).ConfigureAwait(false);
|
|
|
|
var stream = new StreamingCacheWriteStream(
|
|
sourceStream,
|
|
fileStream,
|
|
response,
|
|
bytesWritten => FinalizeStreamingDownload(hash, tempFileName, destinationFilePath, completion, true, bytesWritten),
|
|
bytesWritten => FinalizeStreamingDownload(hash, tempFileName, destinationFilePath, completion, false, bytesWritten),
|
|
_logger);
|
|
|
|
return new CachedFileStreamResult(null, stream, response.Content.Headers.ContentLength);
|
|
}
|
|
|
|
public async Task DownloadFileWhenRequired(string hash)
|
|
{
|
|
var fi = FilePathUtil.GetFileInfoForHash(_hotStoragePath, hash);
|
|
if (fi == null && IsMainServer)
|
|
{
|
|
TryCopyFromColdStorage(hash, FilePathUtil.GetFilePath(_hotStoragePath, hash));
|
|
return;
|
|
}
|
|
|
|
await _downloadSemaphore.WaitAsync().ConfigureAwait(false);
|
|
if ((fi == null || (fi?.Length ?? 0) == 0)
|
|
&& (!_currentTransfers.TryGetValue(hash, out var downloadTask)
|
|
|| (downloadTask?.IsCompleted ?? true)))
|
|
{
|
|
_currentTransfers[hash] = Task.Run(async () =>
|
|
{
|
|
try
|
|
{
|
|
_metrics.IncGauge(MetricsAPI.GaugeFilesDownloadingFromCache);
|
|
await DownloadTask(hash).ConfigureAwait(false);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Error during Download Task for {hash}", hash);
|
|
}
|
|
finally
|
|
{
|
|
_metrics.DecGauge(MetricsAPI.GaugeFilesDownloadingFromCache);
|
|
_currentTransfers.Remove(hash, out _);
|
|
}
|
|
});
|
|
}
|
|
_downloadSemaphore.Release();
|
|
}
|
|
|
|
public FileInfo? GetLocalFilePath(string hash)
|
|
{
|
|
var fi = FilePathUtil.GetFileInfoForHash(_hotStoragePath, hash);
|
|
if (fi == null) return null;
|
|
|
|
try
|
|
{
|
|
File.SetLastAccessTimeUtc(fi.FullName, DateTime.UtcNow);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Failed to update LastAccessTimeUtc for file {hash}", hash);
|
|
}
|
|
|
|
_fileStatisticsService.LogFile(hash, fi.Length);
|
|
|
|
return new FileInfo(fi.FullName);
|
|
}
|
|
|
|
public async Task<FileInfo?> DownloadAndGetLocalFileInfo(string hash)
|
|
{
|
|
await DownloadFileWhenRequired(hash).ConfigureAwait(false);
|
|
|
|
if (_currentTransfers.TryGetValue(hash, out var downloadTask))
|
|
{
|
|
try
|
|
{
|
|
using CancellationTokenSource cts = new();
|
|
cts.CancelAfter(TimeSpan.FromSeconds(300));
|
|
_metrics.IncGauge(MetricsAPI.GaugeFilesTasksWaitingForDownloadFromCache);
|
|
await downloadTask.WaitAsync(cts.Token).ConfigureAwait(false);
|
|
}
|
|
catch (Exception e)
|
|
{
|
|
_logger.LogWarning(e, "Failed while waiting for download task for {hash}", hash);
|
|
return null;
|
|
}
|
|
finally
|
|
{
|
|
_metrics.DecGauge(MetricsAPI.GaugeFilesTasksWaitingForDownloadFromCache);
|
|
}
|
|
}
|
|
|
|
return GetLocalFilePath(hash);
|
|
}
|
|
|
|
public bool AnyFilesDownloading(List<string> hashes)
|
|
{
|
|
return hashes.Exists(_currentTransfers.Keys.Contains);
|
|
}
|
|
|
|
private void FinalizeStreamingDownload(string hash, string? tempFileName, string destinationFilePath,
|
|
TaskCompletionSource<bool> completion, bool success, long bytesWritten)
|
|
{
|
|
try
|
|
{
|
|
if (success)
|
|
{
|
|
if (!string.IsNullOrEmpty(tempFileName))
|
|
{
|
|
File.Move(tempFileName, destinationFilePath, true);
|
|
}
|
|
|
|
var fi = FilePathUtil.GetFileInfoForHash(_hotStoragePath, hash);
|
|
if (fi != null)
|
|
{
|
|
_metrics.IncGauge(MetricsAPI.GaugeFilesTotal);
|
|
_metrics.IncGauge(MetricsAPI.GaugeFilesTotalSize, fi.Length);
|
|
_fileStatisticsService.LogFile(hash, fi.Length);
|
|
_inventoryReporter.ReportAdded(hash);
|
|
}
|
|
}
|
|
else if (!string.IsNullOrEmpty(tempFileName))
|
|
{
|
|
try { File.Delete(tempFileName); } catch { /* ignore */ }
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Failed to finalize streaming download for {hash} after {bytes} bytes", hash, bytesWritten);
|
|
}
|
|
finally
|
|
{
|
|
_metrics.DecGauge(MetricsAPI.GaugeFilesDownloadingFromCache);
|
|
_currentTransfers.Remove(hash, out _);
|
|
completion.TrySetResult(success);
|
|
}
|
|
}
|
|
|
|
private sealed class StreamingCacheWriteStream : Stream
|
|
{
|
|
private readonly Stream _source;
|
|
private readonly FileStream _cacheStream;
|
|
private readonly HttpResponseMessage _response;
|
|
private readonly Action<long> _onSuccess;
|
|
private readonly Action<long> _onFailure;
|
|
private readonly ILogger _logger;
|
|
private long _bytesWritten;
|
|
private int _completed;
|
|
|
|
public StreamingCacheWriteStream(Stream source, FileStream cacheStream, HttpResponseMessage response,
|
|
Action<long> onSuccess, Action<long> onFailure, ILogger logger)
|
|
{
|
|
_source = source;
|
|
_cacheStream = cacheStream;
|
|
_response = response;
|
|
_onSuccess = onSuccess;
|
|
_onFailure = onFailure;
|
|
_logger = logger;
|
|
}
|
|
|
|
public override bool CanRead => true;
|
|
public override bool CanSeek => false;
|
|
public override bool CanWrite => false;
|
|
public override long Length => _bytesWritten;
|
|
public override long Position
|
|
{
|
|
get => _bytesWritten;
|
|
set => throw new NotSupportedException();
|
|
}
|
|
|
|
public override int Read(byte[] buffer, int offset, int count)
|
|
{
|
|
if (Volatile.Read(ref _completed) != 0)
|
|
{
|
|
return 0;
|
|
}
|
|
|
|
try
|
|
{
|
|
int bytesRead = _source.Read(buffer, offset, count);
|
|
if (bytesRead > 0)
|
|
{
|
|
_cacheStream.Write(buffer, offset, bytesRead);
|
|
_bytesWritten += bytesRead;
|
|
return bytesRead;
|
|
}
|
|
|
|
Complete(true);
|
|
return 0;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Streaming download failed while reading");
|
|
Complete(false);
|
|
throw;
|
|
}
|
|
}
|
|
|
|
public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
|
|
{
|
|
if (Volatile.Read(ref _completed) != 0)
|
|
{
|
|
return 0;
|
|
}
|
|
|
|
try
|
|
{
|
|
int bytesRead = await _source.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
|
|
if (bytesRead > 0)
|
|
{
|
|
await _cacheStream.WriteAsync(buffer.Slice(0, bytesRead), cancellationToken).ConfigureAwait(false);
|
|
_bytesWritten += bytesRead;
|
|
return bytesRead;
|
|
}
|
|
|
|
Complete(true);
|
|
return 0;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Streaming download failed while reading");
|
|
Complete(false);
|
|
throw;
|
|
}
|
|
}
|
|
|
|
public override void Flush()
|
|
{
|
|
// no-op
|
|
}
|
|
|
|
protected override void Dispose(bool disposing)
|
|
{
|
|
if (disposing)
|
|
{
|
|
Complete(false);
|
|
}
|
|
|
|
base.Dispose(disposing);
|
|
}
|
|
|
|
private void Complete(bool success)
|
|
{
|
|
if (Interlocked.Exchange(ref _completed, 1) != 0)
|
|
{
|
|
return;
|
|
}
|
|
|
|
try { _cacheStream.Flush(); } catch { /* ignore */ }
|
|
try { _cacheStream.Dispose(); } catch { /* ignore */ }
|
|
try { _source.Dispose(); } catch { /* ignore */ }
|
|
try { _response.Dispose(); } catch { /* ignore */ }
|
|
|
|
if (success)
|
|
{
|
|
_onSuccess(_bytesWritten);
|
|
}
|
|
else
|
|
{
|
|
_onFailure(_bytesWritten);
|
|
}
|
|
}
|
|
|
|
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
|
|
public override void SetLength(long value) => throw new NotSupportedException();
|
|
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
|
|
}
|
|
}
|
|
|
|
public readonly record struct CachedFileStreamResult(FileInfo? File, Stream? Stream, long? ContentLength); |