feat: Batched dependency management

Adaptors/Memory/src/ResultTable.cs

@@ -90,18 +90,17 @@ public Task Create(IEnumerable<Result> results,
   }
 
   /// <inheritdoc />
-  public Task AddTaskDependency(string              sessionId,
-                                ICollection<string> resultIds,
-                                ICollection<string> taskIds,
-                                CancellationToken   cancellationToken)
+  public Task AddTaskDependencies(string                                   sessionId,
+                                  IDictionary<string, ICollection<string>> dependencies,
+                                  CancellationToken                        cancellationToken = default)
   {
     if (!results_.TryGetValue(sessionId,
                               out var session))
     {
-      throw new SessionNotFoundException($"Session '{session}' not found");
+      throw new SessionNotFoundException($"Session '{sessionId}' not found");
     }
 
-    foreach (var resultId in resultIds)
+    foreach (var (resultId, taskIds) in dependencies)
     {
       if (!session.TryGetValue(resultId,
                                out var result))

Adaptors/MongoDB/src/ResultTable.cs

@@ -86,26 +86,36 @@ public async Task Create(IEnumerable<Result> results,
   }
 
   /// <inheritdoc />
-  public async Task AddTaskDependency(string              sessionId,
-                                      ICollection<string> resultIds,
-                                      ICollection<string> taskIds,
-                                      CancellationToken   cancellationToken)
+  public async Task AddTaskDependencies(string                                   sessionId,
+                                        IDictionary<string, ICollection<string>> dependencies,
+                                        CancellationToken                        cancellationToken = default)
   {
-    using var activity = activitySource_.StartActivity($"{nameof(AddTaskDependency)}");
-    activity?.SetTag($"{nameof(AddTaskDependency)}_sessionId",
+    using var activity = activitySource_.StartActivity($"{nameof(AddTaskDependencies)}");
+    activity?.SetTag($"{nameof(AddTaskDependencies)}_sessionId",
                      sessionId);
 
     var resultCollection = resultCollectionProvider_.Get();
 
-    var result = await resultCollection.UpdateManyAsync(Builders<Result>.Filter.Where(model => resultIds.Contains(model.ResultId)),
-                                                        Builders<Result>.Update.AddToSetEach(model => model.DependentTasks,
-                                                                                             taskIds),
-                                                        cancellationToken: cancellationToken)
-                                       .ConfigureAwait(false);
+    if (!dependencies.Any())
+    {
+      return;
+    }
+
+    var dependencyRequests = dependencies.Select(dependency => new UpdateManyModel<Result>(Builders<Result>.Filter.Where(result => dependency.Key == result.ResultId),
+                                                                                           Builders<Result>.Update.AddToSetEach(result => result.DependentTasks,
+                                                                                                                                dependency.Value)));
+
+    var writeResult = await resultCollection.BulkWriteAsync(dependencyRequests,
+                                                            new BulkWriteOptions
+                                                            {
+                                                              IsOrdered = false,
+                                                            },
+                                                            cancellationToken)
+                                            .ConfigureAwait(false);
 
-    if (result.ModifiedCount != resultIds.Count)
+    if (writeResult.ModifiedCount != dependencies.Count)
     {
-      throw new ResultNotFoundException("One of the input result was not found");
+      throw new ResultNotFoundException($"One of the input result was not found: expected: {dependencies.Count}, found: {writeResult.ModifiedCount}");
     }
   }
 

Common/src/Storage/IResultTable.cs

@@ -69,16 +69,14 @@ Task Create(IEnumerable<Result> results,
   ///   Add the tasks Ids to the list of reverse dependencies of the given results
   /// </summary>
   /// <param name="sessionId">Id of the session containing the result</param>
-  /// <param name="resultIds">List of result Id to update</param>
-  /// <param name="taskIds">List of task Id to add to each result dependents</param>
+  /// <param name="dependencies">Dictionary of the dependant tasks for each result</param>
   /// <param name="cancellationToken">Token used to cancel the execution of the method</param>
   /// <returns>
   ///   Task representing the asynchronous execution of the method
   /// </returns>
-  Task AddTaskDependency(string              sessionId,
-                         ICollection<string> resultIds,
-                         ICollection<string> taskIds,
-                         CancellationToken   cancellationToken = default);
+  Task AddTaskDependencies(string                                   sessionId,
+                           IDictionary<string, ICollection<string>> dependencies,
+                           CancellationToken                        cancellationToken = default);
 
   /// <summary>
   ///   Delete the results from the database