Public Access
Sync: mirror a pulled page in one write and read the outbox once per page, not twice per change
This commit is contained in:
1 parent
f997e87c5f
commit
873405e591
4 files changed
+173
-13
No files matched your search
@@ -118,6 +118,65 @@ public sealed class ItemStore(
|
||||
await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Writes the server's version of many items at once, in one transaction, in the order given.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <see cref="SaveAsync"/> for a page of pulled changes. A pull used to write each change on its own —
|
||||
/// a context, a read, a commit — so a first sync of a keychain with a few thousand items was a few
|
||||
/// thousand SQLite transactions, each one a sync to disk. This is one read for the rows that exist
|
||||
/// already and one commit for the lot.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The same blind overwrite, and the same answer when one item appears more than once: they are applied
|
||||
/// in order to the same row, so the last one given is the one kept — exactly what writing them one by one
|
||||
/// would have left.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public async Task SaveManyAsync(IReadOnlyCollection<StoredItem> items, CancellationToken cancellationToken)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(items);
|
||||
|
||||
if (items.Count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var vaultIds = items.Select(item => item.VaultId).Distinct().ToList();
|
||||
var entityIds = items.Select(item => item.EntityId).Distinct().ToList();
|
||||
|
||||
var context = contexts.CreateDbContext();
|
||||
await using var scope = context.ConfigureAwait(false);
|
||||
|
||||
var existing = await context.Set<CachedItemRow>()
|
||||
.Where(r => vaultIds.Contains(r.VaultId) && entityIds.Contains(r.EntityId))
|
||||
.ToListAsync(cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
var rows = existing.ToDictionary(row => (row.VaultId, row.EntityType, row.EntityId));
|
||||
|
||||
foreach (var item in items)
|
||||
{
|
||||
if (!rows.TryGetValue((item.VaultId, item.EntityType, item.EntityId), out var row))
|
||||
{
|
||||
row = new CachedItemRow
|
||||
{
|
||||
VaultId = item.VaultId,
|
||||
EntityType = item.EntityType,
|
||||
EntityId = item.EntityId,
|
||||
};
|
||||
|
||||
context.Add(row);
|
||||
rows.Add((item.VaultId, item.EntityType, item.EntityId), row);
|
||||
}
|
||||
|
||||
Apply(row, item);
|
||||
}
|
||||
|
||||
await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>Removes a tombstone whose change has been seen by everything that needed it.</summary>
|
||||
/// <remarks>
|
||||
/// Only ever called for a row that is already a tombstone. Collecting a live item here would make
|
||||
|
||||
@@ -546,16 +546,25 @@ internal static class ItemMirror
|
||||
Guid vaultId,
|
||||
SyncChange change,
|
||||
CancellationToken cancellationToken) =>
|
||||
items.SaveAsync(
|
||||
new StoredItem(
|
||||
vaultId,
|
||||
change.EntityType,
|
||||
change.EntityId,
|
||||
change.Version,
|
||||
change.ChangeSequence,
|
||||
change.Payload,
|
||||
change.PlaintextFields,
|
||||
change.Operation == SyncOperation.Delete,
|
||||
change.UpdatedAt),
|
||||
cancellationToken);
|
||||
items.SaveAsync(ToStored(vaultId, change), cancellationToken);
|
||||
|
||||
/// <summary>Mirrors a run of changes in one write; see <see cref="ItemStore.SaveManyAsync"/>.</summary>
|
||||
internal static Task WriteManyAsync(
|
||||
ItemStore items,
|
||||
Guid vaultId,
|
||||
IReadOnlyList<SyncChange> changes,
|
||||
CancellationToken cancellationToken) =>
|
||||
items.SaveManyAsync(changes.Select(change => ToStored(vaultId, change)).ToList(), cancellationToken);
|
||||
|
||||
private static StoredItem ToStored(Guid vaultId, SyncChange change) =>
|
||||
new(
|
||||
vaultId,
|
||||
change.EntityType,
|
||||
change.EntityId,
|
||||
change.Version,
|
||||
change.ChangeSequence,
|
||||
change.Payload,
|
||||
change.PlaintextFields,
|
||||
change.Operation == SyncOperation.Delete,
|
||||
change.UpdatedAt);
|
||||
}
|
||||
@@ -154,9 +154,10 @@ public sealed class SyncEngine
|
||||
continue;
|
||||
}
|
||||
|
||||
await ApplyPageAsync(vaultId, response.Changes, report, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
foreach (var change in response.Changes)
|
||||
{
|
||||
await ApplyAsync(vaultId, change, report, cancellationToken).ConfigureAwait(false);
|
||||
report.Pulled++;
|
||||
|
||||
// Counted apart for the same reason the pushed ones are: a machine reads back the log
|
||||
@@ -310,6 +311,61 @@ public sealed class SyncEngine
|
||||
private static string PulledTypes { get; } =
|
||||
string.Join(',', ItemKinds.SyncedTypes.Select(type => (int)type).Order());
|
||||
|
||||
/// <summary>Applies one page of pulled changes, in order, with as few writes as that allows.</summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Each change used to cost a write to mirror it and a read to look for a local edit of the same item —
|
||||
/// two or more SQLite transactions apiece, which on a first sync of a large keychain was most of the time
|
||||
/// the sync took. The outbox is read once per page instead, and every run of changes with no local edit
|
||||
/// waiting is mirrored in a single write.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// ◆ A change that does have an edit waiting goes through <see cref="ApplyAsync"/> exactly as before, with
|
||||
/// the run in front of it written first: the reconciler reads the mirror and the outbox as they stand at
|
||||
/// that change, and both are what the one-by-one loop would have left them as. The one-read snapshot only
|
||||
/// decides which path a change takes; the path that matters re-reads the outbox for itself.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
private async Task ApplyPageAsync(
|
||||
Guid vaultId,
|
||||
IReadOnlyList<SyncChange> changes,
|
||||
SyncReportBuilder report,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (changes.Count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var waiting = (await outbox.ListAllAsync(vaultId, cancellationToken).ConfigureAwait(false))
|
||||
.Select(pending => (pending.EntityType, pending.EntityId))
|
||||
.ToHashSet();
|
||||
|
||||
var run = new List<SyncChange>();
|
||||
|
||||
foreach (var change in changes)
|
||||
{
|
||||
if (!reconcilers.ContainsKey(change.EntityType))
|
||||
{
|
||||
// See ApplyAsync: a type this build cannot sync is neither mirrored nor reconciled.
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!waiting.Contains((change.EntityType, change.EntityId)))
|
||||
{
|
||||
run.Add(change);
|
||||
continue;
|
||||
}
|
||||
|
||||
await ItemMirror.WriteManyAsync(items, vaultId, run, cancellationToken).ConfigureAwait(false);
|
||||
run.Clear();
|
||||
|
||||
await ApplyAsync(vaultId, change, report, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
await ItemMirror.WriteManyAsync(items, vaultId, run, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task ApplyAsync(
|
||||
Guid vaultId,
|
||||
SyncChange change,
|
||||
|
||||
@@ -128,6 +128,42 @@ public sealed class CacheStoreTests : IAsyncLifetime
|
||||
items.ShouldHaveSingleItem().Version.ShouldBe(2);
|
||||
}
|
||||
|
||||
/// <remarks>
|
||||
/// A pull's page in one write. What it leaves has to be exactly what the same items saved one at a time
|
||||
/// would have left: rows that existed replaced, new ones created, and an item given twice ending at the
|
||||
/// later of the two — a page can carry several changes to the same item.
|
||||
/// </remarks>
|
||||
[Fact]
|
||||
public async Task SavingMany_LeavesWhatSavingThemOneAtATimeWould()
|
||||
{
|
||||
var existing = Guid.CreateVersion7();
|
||||
var fresh = Guid.CreateVersion7();
|
||||
var twice = Guid.CreateVersion7();
|
||||
|
||||
await harness.Items.SaveAsync(Item(existing, version: 1, seed: 1), Token);
|
||||
|
||||
await harness.Items.SaveManyAsync(
|
||||
[
|
||||
Item(existing, version: 2, changeSequence: 2, seed: 2),
|
||||
Item(fresh, version: 1, changeSequence: 3, seed: 3),
|
||||
Item(twice, version: 1, changeSequence: 4, seed: 4),
|
||||
Item(twice, version: 2, changeSequence: 5, deleted: true),
|
||||
],
|
||||
Token);
|
||||
|
||||
(await harness.Items.FindAsync(VaultId, SyncEntityType.Host, existing, Token))
|
||||
.ShouldNotBeNull().Version.ShouldBe(2);
|
||||
|
||||
(await harness.Items.FindAsync(VaultId, SyncEntityType.Host, fresh, Token))
|
||||
.ShouldNotBeNull().Version.ShouldBe(1);
|
||||
|
||||
var last = (await harness.Items.FindAsync(VaultId, SyncEntityType.Host, twice, Token)).ShouldNotBeNull();
|
||||
last.Version.ShouldBe(2);
|
||||
last.IsDeleted.ShouldBeTrue();
|
||||
|
||||
(await harness.Items.ListAsync(VaultId, SyncEntityType.Host, true, Token)).Count.ShouldBe(3);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ATombstone_IsHiddenFromTheListButStillFindable()
|
||||
{
|
||||
|
||||
Reference in new issue
Block a user