Job system rewrite

.config/dotnet-tools.json

@@ -3,7 +3,7 @@
   "isRoot": true,
   "tools": {
     "dotnet-ef": {
-      "version": "9.0.9",
+      "version": "9.0.10",
       "commands": [
         "dotnet-ef"
       ],

README.md

@@ -9,10 +9,6 @@ Web application that archives content from online video platforms (currently onl
 Started as a university homework project.
 
 ## DB
-NB!  
-Postgres' max_prepared_transactions option must be set to higher than the default 0.  
-Preferably at least one per connection?
-
 Start local dev DB:
 `docker compose up db -d`
 

RescueTube.Core/BuilderExtensions.cs

@@ -5,8 +5,8 @@
 using RescueTube.Core.DataFetches;
 using RescueTube.Core.JobOrchestration;
 using RescueTube.Core.Jobs;
-using RescueTube.Core.Jobs.Registration;
 using RescueTube.Core.Services;
+using RescueTube.Core.Services.Startup;
 using RescueTube.Core.Utils;
 using RescueTube.Core.Utils.Validation;
 using RescueTube.Domain.Enums;
@@ -20,16 +20,16 @@ public static IServiceCollection AddBll(this IServiceCollection services)
         services.AddOptionsFull<AppPathOptions>(AppPathOptions.Section);
         services.AddSingleton<AppPaths>();
 
-        services.AddSingleton<JobExecutionRegistry>();
         services.AddOptions<JobsConfiguration>();
 
         services.AddOptions<ServiceRegistry>();
 
-        services.AddOptions<DataFetchJobsConfiguration>();
         services.AddScoped<DataFetchService>();
 
         services.AddScoped<ServiceUow>();
 
+        services.AddMemoryCache();
+
         services.AddScoped<SubmissionService>();
         services.AddScoped<ImageService>();
         services.AddScoped<VideoPresentationService>();
@@ -52,18 +52,16 @@ public static IServiceCollection AddBll(this IServiceCollection services)
 
         services.AddScoped<UpdateImagesResolutionJob>();
         services.Configure<JobsConfiguration>(c => c.RegisterJobs(
-            new JobDefinition<DownloadImageJob>
-            {
-                Priority = -1,
-                PreferredMaxConcurrentExecutions = 10,
-            }
+            DownloadImageJob.JobDefinition,
+            DownloadVideoJob.JobDefinition,
+            HandleNextSubmissionJob.JobDefinition,
+            DeleteExpiredRefreshTokensJob.JobDefinition,
+            UpdateImagesResolutionJob.JobDefinition,
+            DataFetchesKillSwitchJob.JobDefinition
         ));
 
-        services.AddOptions<HangfireRecurringJobRegistry>();
-        services.RegisterBllRecurringJobs();
-
-        services.AddScoped<RecurringJobsService>();
-        services.AddHostedService<RegisterRecurringJobsService>();
+        services.AddScoped<IRecurringJobsService, RecurringJobsService>();
+        services.AddHostedService<SetupRecurringJobsService>();
 
         return services;
     }

RescueTube.Core/DTO/JobDefinitionWithSettings.cs

@@ -0,0 +1,16 @@
+using RescueTube.Core.JobOrchestration;
+using RescueTube.Domain.Entities;
+
+namespace RescueTube.Core.DTO;
+
+public record JobDefinitionWithSettings
+{
+    public required JobDefinition JobDefinition { get; init; }
+    public required JobSettings JobSettings { get; init; }
+
+    public void Deconstruct(out JobDefinition jobDefinition, out JobSettings jobSettings)
+    {
+        jobDefinition = JobDefinition;
+        jobSettings = JobSettings;
+    }
+}

RescueTube.Core/Data/AppDbContext.cs

@@ -43,6 +43,7 @@ public abstract class AppDbContext : IdentityDbContext<User, Role, Guid, UserCla
     public DbSet<StatusChangeEvent> StatusChangeEvents => Set<StatusChangeEvent>();
 
     public DbSet<Submission> Submissions => Set<Submission>();
+    public DbSet<SubmissionHandlingFailure> SubmissionHandlingFailures => Set<SubmissionHandlingFailure>();
 
     public DbSet<Playlist> Playlists => Set<Playlist>();
     public DbSet<PlaylistItem> PlaylistItems => Set<PlaylistItem>();
@@ -58,6 +59,8 @@ public abstract class AppDbContext : IdentityDbContext<User, Role, Guid, UserCla
     public DbSet<Setting.String> StringSettings => Set<Setting.String>();
     public DbSet<Setting.DataSize> DataSizeSettings => Set<Setting.DataSize>();
 
+    public DbSet<PersistedJobSettings> JobSettings => Set<PersistedJobSettings>();
+
     private readonly ILoggerFactory? _loggerFactory;
     private readonly DbLoggingOptions? _dbLoggingOptions;
 
@@ -79,6 +82,24 @@ protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
         }
     }
 
+    public void RegisterSavedChangesCallbackRunOnce(Action callback)
+    {
+        RegisterSavedChangesCallbackRunOnce(_ => callback());
+    }
+
+    private void RegisterSavedChangesCallbackRunOnce(Action<object?> callback)
+    {
+        SavedChanges += OnSavedChanges;
+        return;
+
+        void OnSavedChanges(object? sender, EventArgs savedEventArgs)
+        {
+            SavedChanges -= OnSavedChanges;
+
+            callback(sender);
+        }
+    }
+
     public EntityEntry<TEntity> AddIfTracked<TEntity>(TEntity entity) where TEntity : class
     {
         var entry = Entry(entity);

RescueTube.Core/Data/IDataUow.cs

@@ -13,6 +13,4 @@ public interface IDataUow
     public AppDbContext Ctx { get; }
 
     public Task<int> SaveChangesAsync(CancellationToken cancellationToken = default);
-
-    void RegisterSavedChangesCallbackRunOnce(Action callback);
 }

RescueTube.Core/Data/Specifications/IDataFetchSpecification.cs

@@ -8,8 +8,8 @@ public interface IDataFetchSpecification
 {
     public Expression<Func<DataFetch, Video, bool>> IsVideoDataFetch { get; }
     public Expression<Func<DataFetch, bool>> IsOngoing(DateTimeOffset currentTime);
-    public Expression<Func<Author, bool>> ShouldFetchAuthorData(DataFetchJobDefinition jobDefinition);
-    public Expression<Func<Playlist, bool>> ShouldFetchPlaylistData(DataFetchJobDefinition jobDefinition);
+    public Expression<Func<Author, bool>> ShouldFetchAuthorData(DataFetchDefinition dataFetchDefinition, DataFetchJobSettings dataFetchJobSettings);
+    public Expression<Func<Playlist, bool>> ShouldFetchPlaylistData(DataFetchDefinition dataFetchDefinition, DataFetchJobSettings dataFetchJobSettings);
     public Expression<Func<Video, bool>> ShouldFetchVideoData(
-        DataFetchJobDefinition jobDefinition, Expression<Func<Video, bool>> allowRegularFetchesPredicate);
+        DataFetchDefinition dataFetchDefinition, DataFetchJobSettings dataFetchJobSettings, Expression<Func<Video, bool>> allowRegularFetchesPredicate);
 }

RescueTube.Core/DataFetches/DataFetchResult.cs

@@ -0,0 +1,7 @@
+namespace RescueTube.Core.DataFetches;
+
+public enum EntityDataFetchResult
+{
+    Completed,
+    Throttled,
+}

RescueTube.Core/DataFetches/DataFetchService.cs

@@ -1,9 +1,7 @@
-using System.Collections.Concurrent;
 using System.Linq.Expressions;
 using LinqKit;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Logging;
 using RescueTube.Core.Data;
 using RescueTube.Core.Data.Specifications;
 using RescueTube.Domain.Entities;
@@ -17,19 +15,15 @@ public class DataFetchService
     private readonly IDataFetchSpecification _dataFetchSpecification;
     private readonly TimeProvider _timeProvider;
     private readonly IServiceScopeFactory _serviceScopeFactory;
-    private readonly ILogger<DataFetchService> _logger;
 
-    public DataFetchService(AppDbContext dbCtx, IDataFetchSpecification dataFetchSpecification, TimeProvider timeProvider, IServiceScopeFactory serviceScopeFactory, ILogger<DataFetchService> logger)
+    public DataFetchService(AppDbContext dbCtx, IDataFetchSpecification dataFetchSpecification, TimeProvider timeProvider, IServiceScopeFactory serviceScopeFactory)
     {
         _dbCtx = dbCtx;
         _dataFetchSpecification = dataFetchSpecification;
         _timeProvider = timeProvider;
         _serviceScopeFactory = serviceScopeFactory;
-        _logger = logger;
     }
 
-    private static readonly ConcurrentDictionary<DataFetchDefinition, byte> DataFetchStartLocks = [];
-
     public async Task<DataFetchScope?> StartDataFetchAsync(DataFetchDefinition definition, Author author, CancellationToken ct)
     {
         if (definition.EntityType is not EEntityType.Author)
@@ -50,38 +44,28 @@ public DataFetchService(AppDbContext dbCtx, IDataFetchSpecification dataFetchSpe
         return await StartDataFetchAsync(definition, video.Id, ct);
     }
 
-    public async Task<DataFetchScope?> StartDataFetchAsync(DataFetchDefinition definition, Guid? entityId, CancellationToken ct)
+    public async Task<DataFetchScope?> StartDataFetchAsync(DataFetchDefinition definition, Submission submission, CancellationToken ct)
     {
-        // Short-lived lock, used to avoid race condition between IsFetching check and adding new DataFetch.
-        // Not a distributed-safe check, and assumes that this is the only place that adds "Starting" DataFetches.
-        var lockSuccessfullyAcquired = DataFetchStartLocks.TryAdd(definition, byte.MinValue);
-        if (!lockSuccessfullyAcquired)
+        var dataFetch = await AddDataFetchAsync(definition, entityId: null, submission, ct);
+        return new DataFetchScope(dataFetch, this, ct);
+    }
+
+    public async Task<DataFetchScope?> StartDataFetchAsync(DataFetchDefinition definition, Guid entityId, CancellationToken ct)
+    {
+        var hasStartedDataFetch = await IsFetchingAsync(definition, entityId, ct);
+        if (hasStartedDataFetch)
         {
             return null;
         }
 
-        try
-        {
-            var hasStartedDataFetch = entityId is not null && await IsFetchingAsync(definition, entityId.Value, ct);
-            if (hasStartedDataFetch)
-            {
-                return null;
-            }
-
-            var dataFetch = await AddDataFetchAsync(definition, entityId, ct);
-            return new DataFetchScope(dataFetch, this, ct);
-        }
-        finally
-        {
-            var lockSuccessfullyReleased = DataFetchStartLocks.TryRemove(definition, out _);
-            if (!lockSuccessfullyReleased)
-            {
-                _logger.LogError("Failed to release lock for {DataFetchDefinition}, {EntityId}", definition, entityId);
-            }
-        }
+        var dataFetch = await AddDataFetchAsync(definition, entityId, submission: null, ct);
+        return new DataFetchScope(dataFetch, this, ct);
     }
 
-    private async Task<DataFetch> AddDataFetchAsync(DataFetchDefinition definition, Guid? entityId, CancellationToken ct)
+    private async Task<DataFetch> AddDataFetchAsync(
+        DataFetchDefinition definition,
+        Guid? entityId, Submission? submission,
+        CancellationToken ct)
     {
         var now = _timeProvider.GetUtcNow();
 
@@ -94,6 +78,8 @@ private async Task<DataFetch> AddDataFetchAsync(DataFetchDefinition definition,
             LastHeartbeatReceivedAt = now,
             Status = DataFetchStatus.Started,
             DataFetchResults = [],
+            Submission = submission,
+            SubmissionId = submission?.Id,
         };
 
         switch (definition.EntityType)
@@ -131,11 +117,17 @@ await dbCtx.DataFetches
                 .SetProperty(d => d.Message, message)
                 .SetProperty(d => d.StatusUpdatedAt, now));
 
+        var dataFetchEntry = _dbCtx.Entry(dataFetch);
+
         dataFetch.Status = status;
-        _dbCtx.Entry(dataFetch).Property(x => x.Status).IsModified = false;
+        var statusEntry = dataFetchEntry.Property(x => x.Status);
+        statusEntry.OriginalValue = status;
+        statusEntry.IsModified = false;
 
         dataFetch.Message = message;
-        _dbCtx.Entry(dataFetch).Property(x => x.Message).IsModified = false;
+        var messageEntry = _dbCtx.Entry(dataFetch).Property(x => x.Message);
+        messageEntry.OriginalValue = message;
+        messageEntry.IsModified = false;
     }
 
     public async Task SendDataFetchHeartbeatAsync(DataFetch dataFetch, CancellationToken ct)