Retry hangfire jobs

src/SIL.Machine.AspNetCore/Configuration/IMachineBuilderExtensions.cs

@@ -49,6 +49,21 @@ public static IMachineBuilder AddClearMLOptions(this IMachineBuilder builder, IC
         return builder;
     }
 
+    public static IMachineBuilder AddMessageOutboxOptions(
+        this IMachineBuilder builder,
+        Action<MessageOutboxOptions> configureOptions
+    )
+    {
+        builder.Services.Configure(configureOptions);
+        return builder;
+    }
+
+    public static IMachineBuilder AddMessageOutboxOptions(this IMachineBuilder builder, IConfiguration config)
+    {
+        builder.Services.Configure<MessageOutboxOptions>(config);
+        return builder;
+    }
+
     public static IMachineBuilder AddSharedFileOptions(
         this IMachineBuilder builder,
         Action<SharedFileOptions> configureOptions
@@ -219,6 +234,8 @@ public static IMachineBuilder AddMemoryDataAccess(this IMachineBuilder builder)
             o.AddRepository<TranslationEngine>();
             o.AddRepository<RWLock>();
             o.AddRepository<TrainSegmentPair>();
+            o.AddRepository<OutboxMessage>();
+            o.AddRepository<Outbox>();
         });
 
         return builder;
@@ -263,6 +280,14 @@ await c.Indexes.CreateOrUpdateAsync(
                             )
                         )
                 );
+                o.AddRepository<OutboxMessage>(
+                    "outbox_messages",
+                    mapSetup: m => m.MapProperty(m => m.OutboxRef).SetSerializer(new StringSerializer())
+                );
+                o.AddRepository<Outbox>(
+                    "outboxes",
+                    mapSetup: m => m.MapIdProperty(o => o.Id).SetSerializer(new StringSerializer())
+                );
             }
         );
         builder.Services.AddHealthChecks().AddMongoDb(connectionString!, name: "Mongo");
@@ -280,6 +305,11 @@ public static IMachineBuilder AddServalPlatformService(
             throw new InvalidOperationException("Serval connection string is required");
 
         builder.Services.AddScoped<IPlatformService, ServalPlatformService>();
+
+        builder.Services.AddSingleton<IOutboxMessageHandler, ServalPlatformOutboxMessageHandler>();
+
+        builder.Services.AddScoped<IMessageOutboxService, MessageOutboxService>();
+
         builder
             .Services.AddGrpcClient<TranslationPlatformApi.TranslationPlatformApiClient>(o =>
             {
@@ -334,6 +364,7 @@ public static IMachineBuilder AddServalTranslationEngineService(
             options.Interceptors.Add<UnimplementedInterceptor>();
         });
         builder.AddServalPlatformService(connectionString);
+
         engineTypes ??=
             builder.Configuration?.GetSection("TranslationEngines").Get<TranslationEngineType[]?>()
             ?? [TranslationEngineType.SmtTransfer, TranslationEngineType.Nmt];
@@ -397,4 +428,10 @@ public static IMachineBuilder AddModelCleanupService(this IMachineBuilder builde
         builder.Services.AddHostedService<ModelCleanupService>();
         return builder;
     }
+
+    public static IMachineBuilder AddMessageOutboxDeliveryService(this IMachineBuilder builder)
+    {
+        builder.Services.AddHostedService<MessageOutboxDeliveryService>();
+        return builder;
+    }
 }

src/SIL.Machine.AspNetCore/Configuration/IServiceCollectionExtensions.cs

@@ -12,6 +12,7 @@ public static IMachineBuilder AddMachine(this IServiceCollection services, IConf
         services.AddHealthChecks().AddCheck<S3HealthCheck>("S3 Bucket");
 
         services.AddSingleton<ILanguageTagService, LanguageTagService>();
+        services.AddTransient<IFileSystem, FileSystem>();
 
         services.AddScoped<IDistributedReaderWriterLockFactory, DistributedReaderWriterLockFactory>();
         services.AddSingleton<ICorpusService, CorpusService>();
@@ -28,6 +29,7 @@ public static IMachineBuilder AddMachine(this IServiceCollection services, IConf
             builder.AddSmtTransferEngineOptions(o => { });
             builder.AddClearMLOptions(o => { });
             builder.AddBuildJobOptions(o => { });
+            builder.AddMessageOutboxOptions(o => { });
         }
         else
         {
@@ -36,6 +38,7 @@ public static IMachineBuilder AddMachine(this IServiceCollection services, IConf
             builder.AddSmtTransferEngineOptions(configuration.GetSection(SmtTransferEngineOptions.Key));
             builder.AddClearMLOptions(configuration.GetSection(ClearMLOptions.Key));
             builder.AddBuildJobOptions(configuration.GetSection(BuildJobOptions.Key));
+            builder.AddMessageOutboxOptions(configuration.GetSection(MessageOutboxOptions.Key));
         }
         return builder;
     }

src/SIL.Machine.AspNetCore/Configuration/MessageOutboxOptions.cs

@@ -0,0 +1,9 @@
+namespace SIL.Machine.AspNetCore.Configuration;
+
+public class MessageOutboxOptions
+{
+    public const string Key = "MessageOutbox";
+
+    public string OutboxDir { get; set; } = "outbox";
+    public TimeSpan MessageExpirationTimeout { get; set; } = TimeSpan.FromHours(48);
+}

src/SIL.Machine.AspNetCore/Models/Outbox.cs

@@ -0,0 +1,10 @@
+namespace SIL.Machine.AspNetCore.Models;
+
+public record Outbox : IEntity
+{
+    public string Id { get; set; } = "";
+
+    public int Revision { get; set; }
+
+    public int CurrentIndex { get; init; }
+}

src/SIL.Machine.AspNetCore/Models/OutboxMessage.cs

@@ -0,0 +1,15 @@
+namespace SIL.Machine.AspNetCore.Models;
+
+public record OutboxMessage : IEntity
+{
+    public string Id { get; set; } = "";
+    public int Revision { get; set; } = 1;
+    public required int Index { get; init; }
+    public required string OutboxRef { get; init; }
+    public required string Method { get; init; }
+    public required string GroupId { get; init; }
+    public string? Content { get; init; }
+    public required bool HasContentStream { get; init; }
+    public DateTimeOffset Created { get; init; } = DateTimeOffset.UtcNow;
+    public int Attempts { get; init; }
+}

src/SIL.Machine.AspNetCore/Services/ClearMLMonitorService.cs

@@ -83,6 +83,7 @@ await _clearMLService.GetTasksForQueueAsync(_queuePerEngineType[engineType], can
                 _queueSizePerEngineType[engineType] = queuePositionsPerEngineType.Count;
             }
 
+            var dataAccessContext = scope.ServiceProvider.GetRequiredService<IDataAccessContext>();
             var platformService = scope.ServiceProvider.GetRequiredService<IPlatformService>();
             var lockFactory = scope.ServiceProvider.GetRequiredService<IDistributedReaderWriterLockFactory>();
             foreach (TranslationEngine engine in trainingEngines)
@@ -117,6 +118,7 @@ or ClearMLTaskStatus.Completed
                     )
                     {
                         bool canceled = !await TrainJobStartedAsync(
+                            dataAccessContext,
                             lockFactory,
                             buildJobService,
                             platformService,
@@ -169,6 +171,7 @@ await UpdateTrainJobStatus(
                             if (canceling)
                             {
                                 await TrainJobCanceledAsync(
+                                    dataAccessContext,
                                     lockFactory,
                                     buildJobService,
                                     platformService,
@@ -183,6 +186,7 @@ await TrainJobCanceledAsync(
                         case ClearMLTaskStatus.Stopped:
                         {
                             await TrainJobCanceledAsync(
+                                dataAccessContext,
                                 lockFactory,
                                 buildJobService,
                                 platformService,
@@ -196,6 +200,7 @@ await TrainJobCanceledAsync(
                         case ClearMLTaskStatus.Failed:
                         {
                             await TrainJobFaultedAsync(
+                                dataAccessContext,
                                 lockFactory,
                                 buildJobService,
                                 platformService,
@@ -217,6 +222,7 @@ await TrainJobFaultedAsync(
     }
 
     private async Task<bool> TrainJobStartedAsync(
+        IDataAccessContext dataAccessContext,
         IDistributedReaderWriterLockFactory lockFactory,
         IBuildJobService buildJobService,
         IPlatformService platformService,
@@ -225,17 +231,24 @@ private async Task<bool> TrainJobStartedAsync(
         CancellationToken cancellationToken = default
     )
     {
+        bool success;
         IDistributedReaderWriterLock @lock = await lockFactory.CreateAsync(engineId, cancellationToken);
         await using (await @lock.WriterLockAsync(cancellationToken: cancellationToken))
         {
-            if (!await buildJobService.BuildJobStartedAsync(engineId, buildId, cancellationToken))
-                return false;
+            success = await dataAccessContext.WithTransactionAsync(
+                async (ct) =>
+                {
+                    if (!await buildJobService.BuildJobStartedAsync(engineId, buildId, ct))
+                        return false;
+                    await platformService.BuildStartedAsync(buildId, CancellationToken.None);
+                    return true;
+                },
+                cancellationToken: cancellationToken
+            );
         }
-        await platformService.BuildStartedAsync(buildId, CancellationToken.None);
-
         await UpdateTrainJobStatus(platformService, buildId, new ProgressStatus(0), 0, cancellationToken);
         _logger.LogInformation("Build started ({BuildId})", buildId);
-        return true;
+        return success;
     }
 
     private async Task<bool> TrainJobCompletedAsync(
@@ -272,6 +285,7 @@ CancellationToken cancellationToken
     }
 
     private async Task TrainJobFaultedAsync(
+        IDataAccessContext dataAccessContext,
         IDistributedReaderWriterLockFactory lockFactory,
         IBuildJobService buildJobService,
         IPlatformService platformService,
@@ -286,12 +300,18 @@ CancellationToken cancellationToken
             IDistributedReaderWriterLock @lock = await lockFactory.CreateAsync(engineId, cancellationToken);
             await using (await @lock.WriterLockAsync(cancellationToken: cancellationToken))
             {
-                await platformService.BuildFaultedAsync(buildId, message, cancellationToken);
-                await buildJobService.BuildJobFinishedAsync(
-                    engineId,
-                    buildId,
-                    buildComplete: false,
-                    CancellationToken.None
+                await dataAccessContext.WithTransactionAsync(
+                    async (ct) =>
+                    {
+                        await platformService.BuildFaultedAsync(buildId, message, ct);
+                        await buildJobService.BuildJobFinishedAsync(
+                            engineId,
+                            buildId,
+                            buildComplete: false,
+                            CancellationToken.None
+                        );
+                    },
+                    cancellationToken: cancellationToken
                 );
             }
             _logger.LogError("Build faulted ({BuildId}). Error: {ErrorMessage}", buildId, message);
@@ -303,6 +323,7 @@ await buildJobService.BuildJobFinishedAsync(
     }
 
     private async Task TrainJobCanceledAsync(
+        IDataAccessContext dataAccessContext,
         IDistributedReaderWriterLockFactory lockFactory,
         IBuildJobService buildJobService,
         IPlatformService platformService,
@@ -316,12 +337,18 @@ CancellationToken cancellationToken
             IDistributedReaderWriterLock @lock = await lockFactory.CreateAsync(engineId, cancellationToken);
             await using (await @lock.WriterLockAsync(cancellationToken: cancellationToken))
             {
-                await platformService.BuildCanceledAsync(buildId, cancellationToken);
-                await buildJobService.BuildJobFinishedAsync(
-                    engineId,
-                    buildId,
-                    buildComplete: false,
-                    CancellationToken.None
+                await dataAccessContext.WithTransactionAsync(
+                    async (ct) =>
+                    {
+                        await platformService.BuildCanceledAsync(buildId, ct);
+                        await buildJobService.BuildJobFinishedAsync(
+                            engineId,
+                            buildId,
+                            buildComplete: false,
+                            CancellationToken.None
+                        );
+                    },
+                    cancellationToken: cancellationToken
                 );
             }
             _logger.LogInformation("Build canceled ({BuildId})", buildId);

src/SIL.Machine.AspNetCore/Services/FileSystem.cs

@@ -0,0 +1,25 @@
+namespace SIL.Machine.AspNetCore.Services;
+
+public class FileSystem : IFileSystem
+{
+    public void CreateDirectory(string path)
+    {
+        Directory.CreateDirectory(path);
+    }
+
+    public void DeleteFile(string path)
+    {
+        if (File.Exists(path))
+            File.Delete(path);
+    }
+
+    public Stream OpenWrite(string path)
+    {
+        return File.OpenWrite(path);
+    }
+
+    public Stream OpenRead(string path)
+    {
+        return File.OpenRead(path);
+    }
+}