Add Server-Side Download Queue (#21)
* test add queueing to file service * further adjustments to download queueing * add check for whether the request is still in the queue to CheckQueue * forcefully release slot if download didn't finish in 15s * actually cancel the delay task * add metrics and refactor some of the request queue service * refactor pathing * reuse httpclient * add queue request dto to requestfile, enqueue users immediately if a slot is available * change startup to include all controllers * update server pathing * update pathing, again * several adjustments to auth, banning, jwt server tokens, renaming, authorization * update api I guess * adjust automated banning of charaident and reg * generate jwt on servers for internal authentication * remove mvcextensions Co-authored-by: rootdarkarchon <root.darkarchon@outlook.com>
This commit is contained in:
rootdarkarchon
@@ -0,0 +1,102 @@
|
||||
using MareSynchronosShared.Metrics;
|
||||
using MareSynchronosShared.Services;
|
||||
using MareSynchronosStaticFilesServer.Utils;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Net.Http.Headers;
|
||||
using MareSynchronos.API;
|
||||
|
||||
namespace MareSynchronosStaticFilesServer.Services;
|
||||
|
||||
public class CachedFileProvider
|
||||
{
|
||||
private readonly ILogger<CachedFileProvider> _logger;
|
||||
private readonly FileStatisticsService _fileStatisticsService;
|
||||
private readonly MareMetrics _metrics;
|
||||
private readonly Uri _remoteCacheSourceUri;
|
||||
private readonly string _basePath;
|
||||
private readonly ConcurrentDictionary<string, Task> _currentTransfers = new(StringComparer.Ordinal);
|
||||
private readonly HttpClient _httpClient;
|
||||
private bool IsMainServer => _remoteCacheSourceUri == null;
|
||||
|
||||
public CachedFileProvider(IConfigurationService<StaticFilesServerConfiguration> configuration, ILogger<CachedFileProvider> logger, FileStatisticsService fileStatisticsService, MareMetrics metrics)
|
||||
{
|
||||
_logger = logger;
|
||||
_fileStatisticsService = fileStatisticsService;
|
||||
_metrics = metrics;
|
||||
_remoteCacheSourceUri = configuration.GetValueOrDefault<Uri>(nameof(StaticFilesServerConfiguration.RemoteCacheSourceUri), null);
|
||||
_basePath = configuration.GetValue<string>(nameof(StaticFilesServerConfiguration.CacheDirectory));
|
||||
_httpClient = new HttpClient();
|
||||
}
|
||||
|
||||
private async Task DownloadTask(string hash, string auth)
|
||||
{
|
||||
// download file from remote
|
||||
var downloadUrl = MareFiles.ServerFilesGetFullPath(_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(auth);
|
||||
var response = await _httpClient.SendAsync(requestMessage).ConfigureAwait(false);
|
||||
|
||||
try
|
||||
{
|
||||
response.EnsureSuccessStatusCode();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Failed to download {url}", downloadUrl);
|
||||
return;
|
||||
}
|
||||
|
||||
var fileName = FilePathUtil.GetFilePath(_basePath, hash);
|
||||
using var fileStream = new FileStream(fileName, FileMode.OpenOrCreate, FileAccess.ReadWrite);
|
||||
var bufferSize = response.Content.Headers.ContentLength > 1024 * 1024 ? 4096 : 1024;
|
||||
var buffer = new byte[bufferSize];
|
||||
|
||||
var bytesRead = 0;
|
||||
while ((bytesRead = await (await response.Content.ReadAsStreamAsync().ConfigureAwait(false)).ReadAsync(buffer).ConfigureAwait(false)) > 0)
|
||||
{
|
||||
await fileStream.WriteAsync(buffer.AsMemory(0, bytesRead)).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesTotal);
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesTotalSize, FilePathUtil.GetFileInfoForHash(_basePath, hash).Length);
|
||||
}
|
||||
|
||||
public void DownloadFileWhenRequired(string hash, string auth)
|
||||
{
|
||||
var fi = FilePathUtil.GetFileInfoForHash(_basePath, hash);
|
||||
if (fi == null && IsMainServer) return;
|
||||
|
||||
if (fi == null && !_currentTransfers.ContainsKey(hash))
|
||||
{
|
||||
_currentTransfers[hash] = Task.Run(async () =>
|
||||
{
|
||||
await DownloadTask(hash, auth).ConfigureAwait(false);
|
||||
_currentTransfers.Remove(hash, out _);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
public FileStream? GetLocalFileStream(string hash)
|
||||
{
|
||||
var fi = FilePathUtil.GetFileInfoForHash(_basePath, hash);
|
||||
if (fi == null) return null;
|
||||
|
||||
_fileStatisticsService.LogFile(hash, fi.Length);
|
||||
|
||||
return new FileStream(fi.FullName, FileMode.Open, FileAccess.Read, FileShare.Inheritable | FileShare.Read);
|
||||
}
|
||||
|
||||
public async Task<FileStream?> GetAndDownloadFileStream(string hash, string auth)
|
||||
{
|
||||
DownloadFileWhenRequired(hash, auth);
|
||||
|
||||
if (_currentTransfers.TryGetValue(hash, out var downloadTask))
|
||||
{
|
||||
await downloadTask.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return GetLocalFileStream(hash);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,216 @@
|
||||
using ByteSizeLib;
|
||||
using MareSynchronosShared.Data;
|
||||
using MareSynchronosShared.Metrics;
|
||||
using MareSynchronosShared.Models;
|
||||
using MareSynchronosShared.Services;
|
||||
using MareSynchronosStaticFilesServer.Utils;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
|
||||
namespace MareSynchronosStaticFilesServer.Services;
|
||||
|
||||
public class FileCleanupService : IHostedService
|
||||
{
|
||||
private readonly MareMetrics _metrics;
|
||||
private readonly ILogger<FileCleanupService> _logger;
|
||||
private readonly IServiceProvider _services;
|
||||
private readonly IConfigurationService<StaticFilesServerConfiguration> _configuration;
|
||||
private readonly bool _isMainServer;
|
||||
private readonly string _cacheDir;
|
||||
private CancellationTokenSource _cleanupCts;
|
||||
|
||||
public FileCleanupService(MareMetrics metrics, ILogger<FileCleanupService> logger, IServiceProvider services, IConfigurationService<StaticFilesServerConfiguration> configuration)
|
||||
{
|
||||
_metrics = metrics;
|
||||
_logger = logger;
|
||||
_services = services;
|
||||
_configuration = configuration;
|
||||
_isMainServer = configuration.IsMain;
|
||||
_cacheDir = _configuration.GetValue<string>(nameof(StaticFilesServerConfiguration.CacheDirectory));
|
||||
}
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("Cleanup Service started");
|
||||
|
||||
_cleanupCts = new();
|
||||
|
||||
_ = CleanUpTask(_cleanupCts.Token);
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task CleanUpTask(CancellationToken ct)
|
||||
{
|
||||
_logger.LogInformation("Starting periodic cleanup task");
|
||||
|
||||
while (!ct.IsCancellationRequested)
|
||||
{
|
||||
DirectoryInfo dir = new(_cacheDir);
|
||||
var allFiles = dir.GetFiles("*", SearchOption.AllDirectories);
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesTotalSize, allFiles.Sum(f => f.Length));
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesTotal, allFiles.Length);
|
||||
|
||||
using var scope = _services.CreateScope();
|
||||
using var dbContext = scope.ServiceProvider.GetService<MareDbContext>()!;
|
||||
|
||||
await CleanUpOutdatedFiles(dbContext, ct).ConfigureAwait(false);
|
||||
|
||||
CleanUpFilesBeyondSizeLimit(dbContext, ct);
|
||||
|
||||
if (_isMainServer)
|
||||
{
|
||||
await dbContext.SaveChangesAsync(ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
var now = DateTime.Now;
|
||||
TimeOnly currentTime = new(now.Hour, now.Minute, now.Second);
|
||||
TimeOnly futureTime = new(now.Hour, now.Minute - now.Minute % 10, 0);
|
||||
var span = futureTime.AddMinutes(10) - currentTime;
|
||||
|
||||
_logger.LogInformation("File Cleanup Complete, next run at {date}", now.Add(span));
|
||||
await Task.Delay(span, ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
private void CleanUpFilesBeyondSizeLimit(MareDbContext dbContext, CancellationToken ct)
|
||||
{
|
||||
var sizeLimit = _configuration.GetValueOrDefault<double>(nameof(StaticFilesServerConfiguration.CacheSizeHardLimitInGiB), -1);
|
||||
if (sizeLimit <= 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
_logger.LogInformation("Cleaning up files beyond the cache size limit of {cacheSizeLimit} GiB", sizeLimit);
|
||||
var allLocalFiles = Directory.EnumerateFiles(_cacheDir, "*", SearchOption.AllDirectories)
|
||||
.Select(f => new FileInfo(f)).ToList()
|
||||
.OrderBy(f => f.LastAccessTimeUtc).ToList();
|
||||
var totalCacheSizeInBytes = allLocalFiles.Sum(s => s.Length);
|
||||
long cacheSizeLimitInBytes = (long)ByteSize.FromGibiBytes(sizeLimit).Bytes;
|
||||
while (totalCacheSizeInBytes > cacheSizeLimitInBytes && allLocalFiles.Any() && !ct.IsCancellationRequested)
|
||||
{
|
||||
var oldestFile = allLocalFiles[0];
|
||||
allLocalFiles.Remove(oldestFile);
|
||||
totalCacheSizeInBytes -= oldestFile.Length;
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotalSize, oldestFile.Length);
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotal);
|
||||
_logger.LogInformation("Deleting {oldestFile} with size {size}MiB", oldestFile.FullName, ByteSize.FromBytes(oldestFile.Length).MebiBytes);
|
||||
oldestFile.Delete();
|
||||
if (_isMainServer)
|
||||
{
|
||||
FileCache f = new() { Hash = oldestFile.Name.ToUpperInvariant() };
|
||||
dbContext.Entry(f).State = EntityState.Deleted;
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Error during cache size limit cleanup");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task CleanUpOutdatedFiles(MareDbContext dbContext, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
var unusedRetention = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.UnusedFileRetentionPeriodInDays), 14);
|
||||
var forcedDeletionAfterHours = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.ForcedDeletionOfFilesAfterHours), -1);
|
||||
|
||||
_logger.LogInformation("Cleaning up files older than {filesOlderThanDays} days", unusedRetention);
|
||||
if (forcedDeletionAfterHours > 0)
|
||||
{
|
||||
_logger.LogInformation("Cleaning up files written to longer than {hours}h ago", forcedDeletionAfterHours);
|
||||
}
|
||||
|
||||
// clean up files in DB but not on disk or last access is expired
|
||||
var prevTime = DateTime.Now.Subtract(TimeSpan.FromDays(unusedRetention));
|
||||
var prevTimeForcedDeletion = DateTime.Now.Subtract(TimeSpan.FromHours(forcedDeletionAfterHours));
|
||||
var allFiles = await dbContext.Files.ToListAsync().ConfigureAwait(false);
|
||||
int fileCounter = 0;
|
||||
foreach (var fileCache in allFiles.Where(f => f.Uploaded))
|
||||
{
|
||||
var file = FilePathUtil.GetFileInfoForHash(_cacheDir, fileCache.Hash);
|
||||
bool fileDeleted = false;
|
||||
if (file == null && _isMainServer)
|
||||
{
|
||||
_logger.LogInformation("File does not exist anymore: {fileName}", fileCache.Hash);
|
||||
dbContext.Files.Remove(fileCache);
|
||||
fileDeleted = true;
|
||||
}
|
||||
else if (file != null && file.LastAccessTime < prevTime)
|
||||
{
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotalSize, file.Length);
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotal);
|
||||
_logger.LogInformation("File outdated: {fileName}, {fileSize}MiB", file.Name, ByteSize.FromBytes(file.Length).MebiBytes);
|
||||
file.Delete();
|
||||
if (_isMainServer)
|
||||
{
|
||||
fileDeleted = true;
|
||||
dbContext.Files.Remove(fileCache);
|
||||
}
|
||||
}
|
||||
else if (file != null && forcedDeletionAfterHours > 0 && file.LastWriteTime < prevTimeForcedDeletion)
|
||||
{
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotalSize, file.Length);
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotal);
|
||||
_logger.LogInformation("File forcefully deleted: {fileName}, {fileSize}MiB", file.Name, ByteSize.FromBytes(file.Length).MebiBytes);
|
||||
file.Delete();
|
||||
if (_isMainServer)
|
||||
{
|
||||
fileDeleted = true;
|
||||
dbContext.Files.Remove(fileCache);
|
||||
}
|
||||
}
|
||||
|
||||
if (_isMainServer && !fileDeleted && file != null && fileCache.Size == 0)
|
||||
{
|
||||
_logger.LogInformation("Setting File Size of " + fileCache.Hash + " to " + file.Length);
|
||||
fileCache.Size = file.Length;
|
||||
// commit every 1000 files to db
|
||||
if (fileCounter % 1000 == 0) await dbContext.SaveChangesAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
fileCounter++;
|
||||
|
||||
ct.ThrowIfCancellationRequested();
|
||||
}
|
||||
|
||||
// clean up files that are on disk but not in DB for some reason
|
||||
CleanUpOrphanedFiles(allFiles, ct);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Error during file cleanup of old files");
|
||||
}
|
||||
}
|
||||
|
||||
private void CleanUpOrphanedFiles(List<FileCache> allFiles, CancellationToken ct)
|
||||
{
|
||||
if (_isMainServer)
|
||||
{
|
||||
var allFilesHashes = new HashSet<string>(allFiles.Select(a => a.Hash.ToUpperInvariant()), StringComparer.Ordinal);
|
||||
DirectoryInfo dir = new(_cacheDir);
|
||||
var allFilesInDir = dir.GetFiles("*", SearchOption.AllDirectories);
|
||||
foreach (var file in allFilesInDir)
|
||||
{
|
||||
if (!allFilesHashes.Contains(file.Name.ToUpperInvariant()))
|
||||
{
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotalSize, file.Length);
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotal);
|
||||
file.Delete();
|
||||
_logger.LogInformation("File not in DB, deleting: {fileName}", file.Name);
|
||||
}
|
||||
|
||||
ct.ThrowIfCancellationRequested();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_cleanupCts.Cancel();
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
using MareSynchronosShared.Metrics;
|
||||
using System.Collections.Concurrent;
|
||||
|
||||
namespace MareSynchronosStaticFilesServer.Services;
|
||||
|
||||
public class FileStatisticsService : IHostedService
|
||||
{
|
||||
private readonly MareMetrics _metrics;
|
||||
private readonly ILogger<FileStatisticsService> _logger;
|
||||
private CancellationTokenSource _resetCancellationTokenSource;
|
||||
private ConcurrentDictionary<string, long> _pastHourFiles = new(StringComparer.Ordinal);
|
||||
private ConcurrentDictionary<string, long> _pastDayFiles = new(StringComparer.Ordinal);
|
||||
|
||||
public FileStatisticsService(MareMetrics metrics, ILogger<FileStatisticsService> logger)
|
||||
{
|
||||
_metrics = metrics;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public void LogFile(string fileHash, long length)
|
||||
{
|
||||
if (!_pastHourFiles.ContainsKey(fileHash))
|
||||
{
|
||||
_pastHourFiles[fileHash] = length;
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesUniquePastHour);
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesUniquePastHourSize, length);
|
||||
}
|
||||
if (!_pastDayFiles.ContainsKey(fileHash))
|
||||
{
|
||||
_pastDayFiles[fileHash] = length;
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesUniquePastDay);
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesUniquePastDaySize, length);
|
||||
}
|
||||
}
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("Starting FileStatisticsService");
|
||||
_resetCancellationTokenSource = new();
|
||||
_ = ResetHourlyFileData();
|
||||
_ = ResetDailyFileData();
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task ResetHourlyFileData()
|
||||
{
|
||||
while (!_resetCancellationTokenSource.Token.IsCancellationRequested)
|
||||
{
|
||||
_logger.LogInformation("Resetting 1h Data");
|
||||
|
||||
_pastHourFiles = new(StringComparer.Ordinal);
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesUniquePastHour, 0);
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesUniquePastHourSize, 0);
|
||||
|
||||
var now = DateTime.UtcNow;
|
||||
TimeOnly currentTime = new(now.Hour, now.Minute, now.Second);
|
||||
TimeOnly futureTime = new(now.Hour, 0, 0);
|
||||
var span = futureTime.AddHours(1) - currentTime;
|
||||
|
||||
await Task.Delay(span, _resetCancellationTokenSource.Token).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task ResetDailyFileData()
|
||||
{
|
||||
while (!_resetCancellationTokenSource.Token.IsCancellationRequested)
|
||||
{
|
||||
_logger.LogInformation("Resetting 24h Data");
|
||||
|
||||
_pastDayFiles = new(StringComparer.Ordinal);
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesUniquePastDay, 0);
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesUniquePastDaySize, 0);
|
||||
|
||||
var now = DateTime.UtcNow;
|
||||
TimeOnly currentTime = new(now.Hour, now.Minute, now.Second);
|
||||
TimeOnly futureTime = new(0, 0, 0);
|
||||
var span = futureTime - currentTime;
|
||||
|
||||
await Task.Delay(span, _resetCancellationTokenSource.Token).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_resetCancellationTokenSource.Cancel();
|
||||
_logger.LogInformation("Stopping FileStatisticsService");
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,105 @@
|
||||
using Grpc.Core;
|
||||
using MareSynchronosShared.Data;
|
||||
using MareSynchronosShared.Metrics;
|
||||
using MareSynchronosShared.Protos;
|
||||
using MareSynchronosShared.Services;
|
||||
using MareSynchronosStaticFilesServer.Utils;
|
||||
using Microsoft.AspNetCore.Authorization;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
|
||||
namespace MareSynchronosStaticFilesServer.Services;
|
||||
|
||||
[Authorize(Policy = "Internal")]
|
||||
public class GrpcFileService : FileService.FileServiceBase
|
||||
{
|
||||
private readonly string _basePath;
|
||||
private readonly MareDbContext _mareDbContext;
|
||||
private readonly ILogger<GrpcFileService> _logger;
|
||||
private readonly MareMetrics _metricsClient;
|
||||
|
||||
public GrpcFileService(MareDbContext mareDbContext, IConfigurationService<StaticFilesServerConfiguration> configuration, ILogger<GrpcFileService> logger, MareMetrics metricsClient)
|
||||
{
|
||||
_basePath = configuration.GetValue<string>(nameof(StaticFilesServerConfiguration.CacheDirectory));
|
||||
_mareDbContext = mareDbContext;
|
||||
_logger = logger;
|
||||
_metricsClient = metricsClient;
|
||||
}
|
||||
|
||||
[Authorize(Policy = "Internal")]
|
||||
public override async Task<Empty> UploadFile(IAsyncStreamReader<UploadFileRequest> requestStream, ServerCallContext context)
|
||||
{
|
||||
_ = await requestStream.MoveNext().ConfigureAwait(false);
|
||||
var uploadMsg = requestStream.Current;
|
||||
var filePath = FilePathUtil.GetFilePath(_basePath, uploadMsg.Hash);
|
||||
using var fileWriter = File.OpenWrite(filePath);
|
||||
var file = await _mareDbContext.Files.SingleOrDefaultAsync(f => f.Hash == uploadMsg.Hash && f.UploaderUID == uploadMsg.Uploader).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
if (file != null)
|
||||
{
|
||||
await fileWriter.WriteAsync(uploadMsg.FileData.ToArray()).ConfigureAwait(false);
|
||||
|
||||
while (await requestStream.MoveNext().ConfigureAwait(false))
|
||||
{
|
||||
await fileWriter.WriteAsync(requestStream.Current.FileData.ToArray()).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
await fileWriter.FlushAsync().ConfigureAwait(false);
|
||||
fileWriter.Close();
|
||||
|
||||
var fileSize = new FileInfo(filePath).Length;
|
||||
file.Uploaded = true;
|
||||
file.Size = fileSize;
|
||||
|
||||
await _mareDbContext.SaveChangesAsync().ConfigureAwait(false);
|
||||
|
||||
_metricsClient.IncGauge(MetricsAPI.GaugeFilesTotal, 1);
|
||||
_metricsClient.IncGauge(MetricsAPI.GaugeFilesTotalSize, fileSize);
|
||||
|
||||
_logger.LogInformation("User {user} uploaded file {hash}", uploadMsg.Uploader, uploadMsg.Hash);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Error during UploadFile");
|
||||
var fileNew = await _mareDbContext.Files.SingleOrDefaultAsync(f => f.Hash == uploadMsg.Hash && f.UploaderUID == uploadMsg.Uploader).ConfigureAwait(false);
|
||||
if (fileNew != null)
|
||||
{
|
||||
_mareDbContext.Files.Remove(fileNew);
|
||||
}
|
||||
|
||||
await _mareDbContext.SaveChangesAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return new Empty();
|
||||
}
|
||||
|
||||
[Authorize(Policy = "Internal")]
|
||||
public override async Task<Empty> DeleteFiles(DeleteFilesRequest request, ServerCallContext context)
|
||||
{
|
||||
foreach (var hash in request.Hash)
|
||||
{
|
||||
try
|
||||
{
|
||||
var fi = FilePathUtil.GetFileInfoForHash(_basePath, hash);
|
||||
var file = await _mareDbContext.Files.SingleOrDefaultAsync(f => f.Hash == hash).ConfigureAwait(false);
|
||||
if (file != null && fi != null)
|
||||
{
|
||||
_mareDbContext.Files.Remove(file);
|
||||
await _mareDbContext.SaveChangesAsync().ConfigureAwait(false);
|
||||
|
||||
_metricsClient.DecGauge(MetricsAPI.GaugeFilesTotal, fi == null ? 0 : 1);
|
||||
_metricsClient.DecGauge(MetricsAPI.GaugeFilesTotalSize, fi?.Length ?? 0);
|
||||
|
||||
fi?.Delete();
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Could not delete file for hash {hash}", hash);
|
||||
}
|
||||
}
|
||||
|
||||
return new Empty();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,137 @@
|
||||
using MareSynchronos.API;
|
||||
using MareSynchronosShared.Metrics;
|
||||
using MareSynchronosShared.Services;
|
||||
using MareSynchronosStaticFilesServer.Utils;
|
||||
using System.Collections.Concurrent;
|
||||
|
||||
namespace MareSynchronosStaticFilesServer.Services;
|
||||
|
||||
public class RequestQueueService : IHostedService
|
||||
{
|
||||
private CancellationTokenSource _queueCts = new();
|
||||
private readonly UserQueueEntry[] _userQueueRequests;
|
||||
private readonly ConcurrentQueue<UserRequest> _queue = new();
|
||||
private readonly MareMetrics _metrics;
|
||||
private readonly ILogger<RequestQueueService> _logger;
|
||||
private readonly int _queueExpirationSeconds;
|
||||
private SemaphoreSlim _queueSemaphore = new(1);
|
||||
private SemaphoreSlim _queueProcessingSemaphore = new(1);
|
||||
|
||||
public RequestQueueService(MareMetrics metrics, IConfigurationService<StaticFilesServerConfiguration> configurationService, ILogger<RequestQueueService> logger)
|
||||
{
|
||||
_userQueueRequests = new UserQueueEntry[configurationService.GetValueOrDefault(nameof(StaticFilesServerConfiguration.DownloadQueueSize), 50)];
|
||||
_queueExpirationSeconds = configurationService.GetValueOrDefault(nameof(StaticFilesServerConfiguration.DownloadTimeoutSeconds), 5);
|
||||
_metrics = metrics;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public async Task<QueueStatus> EnqueueUser(UserRequest request)
|
||||
{
|
||||
_logger.LogDebug("Enqueueing req {guid} from {user} for {file}", request.RequestId, request.User, request.FileId);
|
||||
|
||||
if (_queueProcessingSemaphore.CurrentCount == 0)
|
||||
{
|
||||
_queue.Enqueue(request);
|
||||
return QueueStatus.Waiting;
|
||||
}
|
||||
|
||||
await _queueSemaphore.WaitAsync().ConfigureAwait(false);
|
||||
QueueStatus status = QueueStatus.Waiting;
|
||||
var idx = Array.FindIndex(_userQueueRequests, r => r == null);
|
||||
if (idx == -1)
|
||||
{
|
||||
_queue.Enqueue(request);
|
||||
status = QueueStatus.Waiting;
|
||||
}
|
||||
else
|
||||
{
|
||||
DequeueIntoSlot(request, idx);
|
||||
status = QueueStatus.Ready;
|
||||
}
|
||||
_queueSemaphore.Release();
|
||||
|
||||
return status;
|
||||
}
|
||||
|
||||
public bool StillEnqueued(Guid request, string user, out int queuePosition)
|
||||
{
|
||||
var result = _queue.FirstOrDefault(c => c.RequestId == request && string.Equals(c.User, user, StringComparison.Ordinal));
|
||||
if (result != null)
|
||||
{
|
||||
queuePosition = Array.IndexOf(_queue.ToArray(), result);
|
||||
return true;
|
||||
}
|
||||
|
||||
queuePosition = -1;
|
||||
return false;
|
||||
}
|
||||
|
||||
public bool IsActiveProcessing(Guid request, string user, out UserRequest userRequest)
|
||||
{
|
||||
var userQueueRequest = _userQueueRequests.Where(u => u != null)
|
||||
.FirstOrDefault(f => f.UserRequest.RequestId == request && string.Equals(f.UserRequest.User, user, StringComparison.Ordinal));
|
||||
userRequest = userQueueRequest?.UserRequest ?? null;
|
||||
return userQueueRequest != null && userRequest != null && userQueueRequest.ExpirationDate > DateTime.UtcNow;
|
||||
}
|
||||
|
||||
public void FinishRequest(Guid request)
|
||||
{
|
||||
var req = _userQueueRequests.First(f => f.UserRequest.RequestId == request);
|
||||
var idx = Array.IndexOf(_userQueueRequests, req);
|
||||
_logger.LogDebug("Finishing Request {guid}, clearing slot {idx}", request, idx);
|
||||
_userQueueRequests[idx] = null;
|
||||
}
|
||||
|
||||
public void ActivateRequest(Guid request)
|
||||
{
|
||||
_logger.LogDebug("Activating request {guid}", request);
|
||||
_userQueueRequests.First(f => f.UserRequest.RequestId == request).IsActive = true;
|
||||
}
|
||||
|
||||
private async Task ProcessRequestQueue(CancellationToken ct)
|
||||
{
|
||||
while (!ct.IsCancellationRequested)
|
||||
{
|
||||
await _queueProcessingSemaphore.WaitAsync(ct).ConfigureAwait(false);
|
||||
await _queueSemaphore.WaitAsync(ct).ConfigureAwait(false);
|
||||
for (int i = 0; i < _userQueueRequests.Length; i++)
|
||||
{
|
||||
if (_userQueueRequests[i] != null && !_userQueueRequests[i].IsActive && _userQueueRequests[i].ExpirationDate < DateTime.UtcNow) _userQueueRequests[i] = null;
|
||||
|
||||
if (_userQueueRequests[i] == null)
|
||||
{
|
||||
if (_queue.TryDequeue(out var request))
|
||||
{
|
||||
DequeueIntoSlot(request, i);
|
||||
}
|
||||
}
|
||||
|
||||
if (!_queue.Any()) break;
|
||||
}
|
||||
_queueProcessingSemaphore.Release();
|
||||
_queueSemaphore.Release();
|
||||
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeDownloadQueue, _queue.Count);
|
||||
|
||||
await Task.Delay(250).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
private void DequeueIntoSlot(UserRequest userRequest, int slot)
|
||||
{
|
||||
_logger.LogDebug("Dequeueing {req} into {i}: {user} with {file}", userRequest.RequestId, slot, userRequest.User, userRequest.FileId);
|
||||
_userQueueRequests[slot] = new(userRequest, DateTime.UtcNow.AddSeconds(_queueExpirationSeconds));
|
||||
}
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_ = Task.Run(() => ProcessRequestQueue(_queueCts.Token));
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_queueCts.Cancel();
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user