Pull updating commits hashes out of SnapshotWorker

myieye · myieye · commit 0dc12e5b50f1 · 2025-10-24T14:59:15.000+02:00
diff --git a/src/SIL.Harmony.Core/QueryHelpers.cs b/src/SIL.Harmony.Core/QueryHelpers.cs
@@ -60,6 +60,22 @@ public static async IAsyncEnumerable<TCommit> GetMissingCommits<TCommit, TChange
         }
     }
 
+    private static readonly IComparer<CommitBase> CommitComparer =
+        Comparer<CommitBase>.Create((a, b) => a.CompareKey.CompareTo(b.CompareKey));
+
+    public static SortedSet<T> ToSortedSet<T>(this IEnumerable<T> queryable) where T : CommitBase
+    {
+        return new SortedSet<T>(queryable, CommitComparer);
+    }
+
+    public static async Task<SortedSet<T>> ToSortedSetAsync<T>(this IQueryable<T> queryable) where T : CommitBase
+    {
+        var set = new SortedSet<T>(CommitComparer);
+        await foreach (var item in queryable.AsAsyncEnumerable())
+            set.Add(item);
+        return set;
+    }
+
     public static IQueryable<T> DefaultOrder<T>(this IQueryable<T> queryable) where T: CommitBase
     {
         return queryable
diff --git a/src/SIL.Harmony/DataModel.cs b/src/SIL.Harmony/DataModel.cs
@@ -81,8 +81,8 @@ public async Task AddManyChanges(Guid clientId,
         repo.ClearChangeTracker();
 
         await using var transaction = await repo.BeginTransactionAsync();
-        await repo.AddCommits(commits);
-        await UpdateSnapshots(repo, commits.First(), commits);
+        var updatedCommits = await repo.AddCommits(commits);
+        await UpdateSnapshots(repo, updatedCommits);
         await ValidateCommits(repo);
         await transaction.CommitAsync();
     }
@@ -119,8 +119,8 @@ private async Task Add(Commit commit)
         repo.ClearChangeTracker();
 
         await using var transaction = repo.IsInTransaction ? null : await repo.BeginTransactionAsync();
-        await repo.AddCommit(commit);
-        await UpdateSnapshots(repo, commit, [commit]);
+        var updatedCommits = await repo.AddCommit(commit);
+        await UpdateSnapshots(repo, updatedCommits);
 
         if (AlwaysValidate) await ValidateCommits(repo);
 
@@ -156,8 +156,8 @@ async Task ISyncable.AddRangeFromSync(IEnumerable<Commit> commits)
 
             await using var transaction = await repo.BeginTransactionAsync();
             //don't save since UpdateSnapshots will also modify newCommits with hashes, so changes will be saved once that's done
-            await repo.AddCommits(newCommits, false);
-            await UpdateSnapshots(repo, oldestChange, newCommits);
+            var updatedCommits = await repo.AddCommits(newCommits, false);
+            await UpdateSnapshots(repo, updatedCommits);
             await ValidateCommits(repo);
             await transaction.CommitAsync();
         }
@@ -194,13 +194,15 @@ ValueTask<bool> ISyncable.ShouldSync()
         return ValueTask.FromResult(true);
     }
 
-    private async Task UpdateSnapshots(CrdtRepository repo, Commit oldestAddedCommit, Commit[] newCommits)
+    private async Task UpdateSnapshots(CrdtRepository repo, SortedSet<Commit> commitsToApply)
     {
+        if (commitsToApply.Count == 0) return;
+        var oldestAddedCommit = commitsToApply.First();
         await repo.DeleteStaleSnapshots(oldestAddedCommit);
         Dictionary<Guid, Guid?> snapshotLookup;
-        if (newCommits.Length > 10)
+        if (commitsToApply.Count > 10)
         {
-            var entityIds = newCommits.SelectMany(c => c.ChangeEntities.Select(ce => ce.EntityId));
+            var entityIds = commitsToApply.SelectMany(c => c.ChangeEntities.Select(ce => ce.EntityId));
             snapshotLookup = await repo.CurrentSnapshots()
                 .Where(s => entityIds.Contains(s.EntityId))
                 .Select(s => new KeyValuePair<Guid, Guid?>(s.EntityId, s.Id))
@@ -212,7 +214,7 @@ private async Task UpdateSnapshots(CrdtRepository repo, Commit oldestAddedCommit
         }
 
         var snapshotWorker = new SnapshotWorker(snapshotLookup, repo, _crdtConfig.Value);
-        await snapshotWorker.UpdateSnapshots(oldestAddedCommit, newCommits);
+        await snapshotWorker.UpdateSnapshots(commitsToApply);
     }
 
     private async Task ValidateCommits(CrdtRepository repo)
@@ -240,8 +242,8 @@ public async Task RegenerateSnapshots()
         await using var repo = await _crdtRepositoryFactory.CreateRepository();
         await repo.DeleteSnapshotsAndProjectedTables();
         repo.ClearChangeTracker();
-        var allCommits = await repo.CurrentCommits().AsNoTracking().ToArrayAsync();
-        await UpdateSnapshots(repo, allCommits.First(), allCommits);
+        var allCommits = await repo.CurrentCommits().AsNoTracking().ToSortedSetAsync();
+        await UpdateSnapshots(repo, allCommits);
     }
 
     public async Task<ObjectSnapshot> GetLatestSnapshotByObjectId(Guid entityId)
@@ -296,7 +298,7 @@ public async Task<Dictionary<Guid, ObjectSnapshot>> GetSnapshotsAtCommit(Commit
         var repository = repo.GetScopedRepository(commit);
         var (snapshots, pendingCommits) = await repository.GetCurrentSnapshotsAndPendingCommits();
 
-        if (pendingCommits.Length != 0)
+        if (pendingCommits.Count != 0)
         {
             snapshots = await SnapshotWorker.ApplyCommitsToSnapshots(snapshots,
                 repository,
@@ -331,8 +333,8 @@ public async Task<T> GetAtCommit<T>(Commit commit, Guid entityId)
         var newCommits = await repository.CurrentCommits()
             .Include(c => c.ChangeEntities)
             .WhereAfter(snapshot.Commit)
-            .ToArrayAsync();
-        if (newCommits.Length > 0)
+            .ToSortedSetAsync();
+        if (newCommits.Count > 0)
         {
             var snapshots = await SnapshotWorker.ApplyCommitsToSnapshots(
                 new Dictionary<Guid, ObjectSnapshot>([
diff --git a/src/SIL.Harmony/Db/CrdtRepository.cs b/src/SIL.Harmony/Db/CrdtRepository.cs
@@ -207,7 +207,7 @@ public IAsyncEnumerable<SimpleSnapshot> CurrenSimpleSnapshots(bool includeDelete
         return snapshots;
     }
 
-    public async Task<(Dictionary<Guid, ObjectSnapshot> currentSnapshots, Commit[] pendingCommits)> GetCurrentSnapshotsAndPendingCommits()
+    public async Task<(Dictionary<Guid, ObjectSnapshot> currentSnapshots, SortedSet<Commit> pendingCommits)> GetCurrentSnapshotsAndPendingCommits()
     {
         var snapshots = await CurrentSnapshots().Include(s => s.Commit).ToDictionaryAsync(s => s.EntityId);
 
@@ -217,7 +217,7 @@ public IAsyncEnumerable<SimpleSnapshot> CurrenSimpleSnapshots(bool includeDelete
         var newCommits = await CurrentCommits()
             .Include(c => c.ChangeEntities)
             .WhereAfter(lastCommit)
-            .ToArrayAsync();
+            .ToSortedSetAsync();
         return (snapshots, newCommits);
     }
 
@@ -326,23 +326,11 @@ public async Task AddSnapshots(IEnumerable<ObjectSnapshot> snapshots)
             catch (DbUpdateException e)
             {
                 var entries = string.Join(Environment.NewLine, e.Entries.Select(entry => entry.ToString()));
-                var message = $"Error saving snapshots: {e.Message}{Environment.NewLine}{entries}";
+                var message = $"Error saving snapshots (deleted: {grouping.Key}): {e.Message}{Environment.NewLine}{entries}";
                 _logger.LogError(e, message);
                 throw new DbUpdateException(message, e);
             }
         }
-
-        try
-        {
-            await _dbContext.SaveChangesAsync();
-        }
-        catch (DbUpdateException e)
-        {
-            var entries = string.Join(Environment.NewLine, e.Entries.Select(entry => entry.ToString()));
-            var message = $"Error saving snapshots: {e.Message}{Environment.NewLine}{entries}";
-            _logger.LogError(e, message);
-            throw new DbUpdateException(message, e);
-        }
     }
 
     private async ValueTask ProjectSnapshot(ObjectSnapshot objectSnapshot)
@@ -387,16 +375,45 @@ public CrdtRepository GetScopedRepository(Commit excludeChangesAfterCommit)
         return new CrdtRepository(_dbContext, _crdtConfig, _logger, excludeChangesAfterCommit);
     }
 
-    public async Task AddCommit(Commit commit)
+    public async Task<SortedSet<Commit>> AddCommit(Commit commit)
     {
-        _dbContext.Add(commit);
+        var updatedCommits = await AddNewCommits([commit]);
         await _dbContext.SaveChangesAsync();
+        return updatedCommits;
     }
 
-    public async Task AddCommits(IEnumerable<Commit> commits, bool save = true)
+    public async Task<SortedSet<Commit>> AddCommits(IEnumerable<Commit> commits, bool save = true)
     {
-        _dbContext.AddRange(commits);
+        var updatedCommits = await AddNewCommits(commits);
         if (save) await _dbContext.SaveChangesAsync();
+        return updatedCommits;
+    }
+
+    private async Task<SortedSet<Commit>> AddNewCommits(IEnumerable<Commit> newCommits)
+    {
+        if (newCommits is null || !newCommits.Any()) return [];
+        var oldestAddedCommit = newCommits.MinBy(c => c.CompareKey)
+            ?? throw new ArgumentException("Couldn't find oldest commit", nameof(newCommits));
+        var parentCommit = await FindPreviousCommit(oldestAddedCommit);
+        var existingCommitsToUpdate = await GetCommitsAfter(parentCommit);
+        var commitsToApply = existingCommitsToUpdate
+            .UnionBy(newCommits, c => c.Id)
+            .ToSortedSet();
+        //we're inserting commits in the past/rewriting history, so we need to update the previous commit hashes
+        await UpdateCommitHashes(commitsToApply, parentCommit);
+        _dbContext.AddRange(newCommits);
+        return commitsToApply;
+    }
+
+    private async Task UpdateCommitHashes(SortedSet<Commit> commits, Commit? parentCommit = null)
+    {
+        var previousCommitHash = parentCommit?.Hash ?? CommitBase.NullParentHash;
+        foreach (var commit in commits)
+        {
+            commit.SetParentHash(previousCommitHash);
+            previousCommitHash = commit.Hash;
+        }
+        await _dbContext.SaveChangesAsync();
     }
 
     public HybridDateTime? GetLatestDateTime()
diff --git a/src/SIL.Harmony/SnapshotWorker.cs b/src/SIL.Harmony/SnapshotWorker.cs
@@ -31,12 +31,12 @@ private SnapshotWorker(Dictionary<Guid, ObjectSnapshot> snapshots,
     internal static async Task<Dictionary<Guid, ObjectSnapshot>> ApplyCommitsToSnapshots(
         Dictionary<Guid, ObjectSnapshot> snapshots,
         CrdtRepository crdtRepository,
-        ICollection<Commit> commits,
+        SortedSet<Commit> commits,
         CrdtConfig crdtConfig)
     {
         //we need to pass in the snapshots because we expect it to be modified, this is intended.
         //if the constructor makes a copy in the future this will need to be updated
-        await new SnapshotWorker(snapshots, [], crdtRepository, crdtConfig).ApplyCommitChanges(commits, false, null);
+        await new SnapshotWorker(snapshots, [], crdtRepository, crdtConfig).ApplyCommitChanges(commits);
         return snapshots;
     }
 
@@ -49,32 +49,22 @@ internal SnapshotWorker(Dictionary<Guid, Guid?> snapshotLookup,
     {
     }
 
-    public async Task UpdateSnapshots(Commit oldestAddedCommit, Commit[] newCommits)
+    public async Task UpdateSnapshots(SortedSet<Commit> commits)
     {
-        var previousCommit = await _crdtRepository.FindPreviousCommit(oldestAddedCommit);
-        var commits = await _crdtRepository.GetCommitsAfter(previousCommit);
-        await ApplyCommitChanges(commits.UnionBy(newCommits, c => c.Id), true, previousCommit?.Hash ?? CommitBase.NullParentHash);
-
+        await ApplyCommitChanges(commits);
         await _crdtRepository.AddSnapshots([
             .._rootSnapshots.Values,
             .._newIntermediateSnapshots,
             .._pendingSnapshots.Values
         ]);
     }
 
-    private async ValueTask ApplyCommitChanges(IEnumerable<Commit> commits, bool updateCommitHash, string? previousCommitHash)
+    private async ValueTask ApplyCommitChanges(SortedSet<Commit> commits)
     {
         var intermediateSnapshots = new Dictionary<Guid, ObjectSnapshot>();
         var commitIndex = 0;
-        foreach (var commit in commits.DefaultOrder())
+        foreach (var commit in commits)
         {
-            if (updateCommitHash && previousCommitHash is not null)
-            {
-                //we're rewriting history, so we need to update the previous commit hash
-                commit.SetParentHash(previousCommitHash);
-            }
-
-            previousCommitHash = commit.Hash;
             commitIndex++;
             foreach (var commitChange in commit.ChangeEntities.OrderBy(c => c.Index))
             {

Original file line number	Diff line number	Diff line change
`@@ -31,12 +31,12 @@ private SnapshotWorker(Dictionary<Guid, ObjectSnapshot> snapshots,`
`31`	`31`	`internal static async Task<Dictionary<Guid, ObjectSnapshot>> ApplyCommitsToSnapshots(`
`32`	`32`	`Dictionary<Guid, ObjectSnapshot> snapshots,`
`33`	`33`	`CrdtRepository crdtRepository,`
`34`		`- ICollection<Commit> commits,`
	`34`	`+ SortedSet<Commit> commits,`
`35`	`35`	`CrdtConfig crdtConfig)`
`36`	`36`	`{`
`37`	`37`	`//we need to pass in the snapshots because we expect it to be modified, this is intended.`
`38`	`38`	`//if the constructor makes a copy in the future this will need to be updated`
`39`		`- await new SnapshotWorker(snapshots, [], crdtRepository, crdtConfig).ApplyCommitChanges(commits, false, null);`
	`39`	`+ await new SnapshotWorker(snapshots, [], crdtRepository, crdtConfig).ApplyCommitChanges(commits);`
`40`	`40`	`return snapshots;`
`41`	`41`	`}`
`42`	`42`
`@@ -49,32 +49,22 @@ internal SnapshotWorker(Dictionary<Guid, Guid?> snapshotLookup,`
`49`	`49`	`{`
`50`	`50`	`}`
`51`	`51`
`52`		`- public async Task UpdateSnapshots(Commit oldestAddedCommit, Commit[] newCommits)`
	`52`	`+ public async Task UpdateSnapshots(SortedSet<Commit> commits)`
`53`	`53`	`{`
`54`		`- var previousCommit = await _crdtRepository.FindPreviousCommit(oldestAddedCommit);`
`55`		`- var commits = await _crdtRepository.GetCommitsAfter(previousCommit);`
`56`		`- await ApplyCommitChanges(commits.UnionBy(newCommits, c => c.Id), true, previousCommit?.Hash ?? CommitBase.NullParentHash);`
`57`		`-`
	`54`	`+ await ApplyCommitChanges(commits);`
`58`	`55`	`await _crdtRepository.AddSnapshots([`
`59`	`56`	`.._rootSnapshots.Values,`
`60`	`57`	`.._newIntermediateSnapshots,`
`61`	`58`	`.._pendingSnapshots.Values`
`62`	`59`	`]);`
`63`	`60`	`}`
`64`	`61`
`65`		`- private async ValueTask ApplyCommitChanges(IEnumerable<Commit> commits, bool updateCommitHash, string? previousCommitHash)`
	`62`	`+ private async ValueTask ApplyCommitChanges(SortedSet<Commit> commits)`
`66`	`63`	`{`
`67`	`64`	`var intermediateSnapshots = new Dictionary<Guid, ObjectSnapshot>();`
`68`	`65`	`var commitIndex = 0;`
`69`		`- foreach (var commit in commits.DefaultOrder())`
	`66`	`+ foreach (var commit in commits)`
`70`	`67`	`{`
`71`		`- if (updateCommitHash && previousCommitHash is not null)`
`72`		`- {`
`73`		`- //we're rewriting history, so we need to update the previous commit hash`
`74`		`- commit.SetParentHash(previousCommitHash);`
`75`		`- }`
`76`		`-`
`77`		`- previousCommitHash = commit.Hash;`
`78`	`68`	`commitIndex++;`
`79`	`69`	`foreach (var commitChange in commit.ChangeEntities.OrderBy(c => c.Index))`
`80`	`70`	`{`