Public Access
Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
858e36279d | ||
|
|
55e5de0ce4 | ||
|
|
8cb1457ca9 |
No files matched your search
@@ -68,7 +68,15 @@ the Npgsql connection string** — the default; do not enable multiplexing.
|
|||||||
visible conflicted copy for opaque ones. **It must never silently drop a key or a host.**
|
visible conflicted copy for opaque ones. **It must never silently drop a key or a host.**
|
||||||
- Deletes are revisioned tombstones, garbage-collected after 90 days. Sync must therefore be
|
- Deletes are revisioned tombstones, garbage-collected after 90 days. Sync must therefore be
|
||||||
able to read tombstones, which is the one place that legitimately bypasses the soft-delete
|
able to read tombstones, which is the one place that legitimately bypasses the soft-delete
|
||||||
query filter — guarded by an explicit permission check.
|
query filter — guarded by an explicit permission check. **The collection has since been built**
|
||||||
|
(`TombstoneCollector`), and it costs a machine away for longer than the retention its news of
|
||||||
|
those deletions. So a collection raises the vault's sync horizon, and a cursor from before it is
|
||||||
|
refused. The client then reads from the beginning, and when that read finishes it removes the
|
||||||
|
items it held that the read never mentioned. An item with an unsent local edit is kept: the
|
||||||
|
push finds the server without it, puts it back, and records a `RecreatedAfterCollection`
|
||||||
|
conflict. A cursor carries the horizon it was issued under. Without that, the read from the
|
||||||
|
beginning would be refused on its own early pages, which sit behind the horizon whenever an
|
||||||
|
older item survives.
|
||||||
- **An `Infrastructure.Tests` case must prove cursor ordering under N concurrent pushes.**
|
- **An `Infrastructure.Tests` case must prove cursor ordering under N concurrent pushes.**
|
||||||
Without it this ADR's central bug is invisible until production.
|
Without it this ADR's central bug is invisible until production.
|
||||||
- Two concurrency mechanisms, deliberately: `version integer` is the client-visible monotonic
|
- Two concurrency mechanisms, deliberately: `version integer` is the client-visible monotonic
|
||||||
|
|||||||
@@ -27,4 +27,16 @@ internal static partial class SyncLog
|
|||||||
int appliedCount,
|
int appliedCount,
|
||||||
int operationCount,
|
int operationCount,
|
||||||
Guid vaultId);
|
Guid vaultId);
|
||||||
|
|
||||||
|
[LoggerMessage(
|
||||||
|
EventId = 2003,
|
||||||
|
Level = LogLevel.Information,
|
||||||
|
Message = "Collected {Count} expired tombstone(s) and superseded change-log entries.")]
|
||||||
|
internal static partial void Collected(ILogger logger, int count);
|
||||||
|
|
||||||
|
[LoggerMessage(
|
||||||
|
EventId = 2004,
|
||||||
|
Level = LogLevel.Warning,
|
||||||
|
Message = "Tombstone collection failed; it will run again at the next interval.")]
|
||||||
|
internal static partial void CollectionFailed(ILogger logger, Exception exception);
|
||||||
}
|
}
|
||||||
@@ -26,12 +26,59 @@ internal sealed class SyncService(
|
|||||||
CancellationToken cancellationToken)
|
CancellationToken cancellationToken)
|
||||||
{
|
{
|
||||||
var afterSequence = 0L;
|
var afterSequence = 0L;
|
||||||
|
var issuedUnder = 0L;
|
||||||
if (!string.IsNullOrEmpty(request.Cursor)
|
if (!string.IsNullOrEmpty(request.Cursor)
|
||||||
&& !SyncCursor.TryDecode(cursorKeys.Key, request.Cursor, vault.Id, out afterSequence))
|
&& !SyncCursor.TryDecode(cursorKeys.Key, request.Cursor, vault.Id, out afterSequence, out issuedUnder))
|
||||||
{
|
{
|
||||||
throw new InvalidCursorException();
|
throw new InvalidCursorException();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Read before the changes, for the cursor this read hands out; see the check after it.
|
||||||
|
var horizonBefore = await HorizonAsync(vault.Id, cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
|
var (changes, hasMore) = await ReadPageAsync(vault.Id, afterSequence, request, cancellationToken)
|
||||||
|
.ConfigureAwait(false);
|
||||||
|
|
||||||
|
// â—† After the read, not before it. The collector deletes a tombstone's entries and raises the horizon
|
||||||
|
// in one transaction, so a read that already missed an entry is followed by a horizon that says so;
|
||||||
|
// checked first, the two could straddle a collection and the deletion would go unnoticed for good.
|
||||||
|
//
|
||||||
|
// A cursor behind the horizon is refused only if it was issued before it. One issued since belongs to
|
||||||
|
// a read from the beginning that started after the collection, and its early pages are behind the
|
||||||
|
// horizon because old, untouched items are; refusing those would send that read back to the start
|
||||||
|
// for ever. See SyncCursor.
|
||||||
|
var horizon = await HorizonAsync(vault.Id, cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
|
if (afterSequence > 0 && afterSequence < horizon && issuedUnder < horizon)
|
||||||
|
{
|
||||||
|
throw new InvalidCursorException();
|
||||||
|
}
|
||||||
|
|
||||||
|
var hydrated = await HydrateAsync(vault, changes, cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
|
// When nothing came back the cursor must not move, or a concurrent write landing between
|
||||||
|
// this read and the next would be skipped forever.
|
||||||
|
var nextSequence = changes.Count > 0 ? changes[^1].Sequence : afterSequence;
|
||||||
|
|
||||||
|
return new SyncPullResponse(
|
||||||
|
Changes: hydrated,
|
||||||
|
// The horizon from before the read, never the one after: a collection landing between the two
|
||||||
|
// could have taken an entry this page would otherwise have carried, and a cursor claiming the
|
||||||
|
// later horizon would vouch for a client that never heard of it. The earlier one costs at most
|
||||||
|
// a restart.
|
||||||
|
NextCursor: SyncCursor.Encode(cursorKeys.Key, vault.Id, nextSequence, horizonBefore),
|
||||||
|
HasMore: hasMore,
|
||||||
|
ServerTime: clock.GetUtcNow(),
|
||||||
|
CurrentKeyGeneration: (uint)vault.KeyGeneration);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>The changes after a position, one page of them, and whether there are more.</summary>
|
||||||
|
private async Task<(List<VaultChange> Changes, bool HasMore)> ReadPageAsync(
|
||||||
|
Guid vaultId,
|
||||||
|
long afterSequence,
|
||||||
|
SyncPullRequest request,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
var limit = Math.Clamp(
|
var limit = Math.Clamp(
|
||||||
request.Limit ?? options.DefaultPullLimit,
|
request.Limit ?? options.DefaultPullLimit,
|
||||||
1,
|
1,
|
||||||
@@ -43,7 +90,7 @@ internal sealed class SyncService(
|
|||||||
|
|
||||||
// One extra row, so "is there more" needs no second query.
|
// One extra row, so "is there more" needs no second query.
|
||||||
var query = database.VaultChanges
|
var query = database.VaultChanges
|
||||||
.Where(c => c.VaultId == vault.Id && c.Sequence > afterSequence);
|
.Where(c => c.VaultId == vaultId && c.Sequence > afterSequence);
|
||||||
|
|
||||||
if (types is not null)
|
if (types is not null)
|
||||||
{
|
{
|
||||||
@@ -62,18 +109,7 @@ internal sealed class SyncService(
|
|||||||
changes.RemoveAt(changes.Count - 1);
|
changes.RemoveAt(changes.Count - 1);
|
||||||
}
|
}
|
||||||
|
|
||||||
var hydrated = await HydrateAsync(vault, changes, cancellationToken).ConfigureAwait(false);
|
return (changes, hasMore);
|
||||||
|
|
||||||
// When nothing came back the cursor must not move, or a concurrent write landing between
|
|
||||||
// this read and the next would be skipped forever.
|
|
||||||
var nextSequence = changes.Count > 0 ? changes[^1].Sequence : afterSequence;
|
|
||||||
|
|
||||||
return new SyncPullResponse(
|
|
||||||
Changes: hydrated,
|
|
||||||
NextCursor: SyncCursor.Encode(cursorKeys.Key, vault.Id, nextSequence),
|
|
||||||
HasMore: hasMore,
|
|
||||||
ServerTime: clock.GetUtcNow(),
|
|
||||||
CurrentKeyGeneration: (uint)vault.KeyGeneration);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
@@ -622,14 +658,30 @@ internal sealed class SyncService(
|
|||||||
/// <summary>The kind behind a change-log entry, or null for a type this server does not synchronise.</summary>
|
/// <summary>The kind behind a change-log entry, or null for a type this server does not synchronise.</summary>
|
||||||
private static IItemKind? KindForChange(ChangeEntityType type) => ItemKinds.For(ToWire(type));
|
private static IItemKind? KindForChange(ChangeEntityType type) => ItemKinds.For(ToWire(type));
|
||||||
|
|
||||||
private async Task<long> CurrentHeadAsync(Guid vaultId, CancellationToken cancellationToken) =>
|
/// <remarks>
|
||||||
await database.VaultChanges
|
/// Never below the vault's horizon: collecting a tombstone can take the newest entries with it, and a
|
||||||
|
/// head reported lower than a cursor the client already holds would rewind it under the horizon and
|
||||||
|
/// cost a full resync for nothing.
|
||||||
|
/// </remarks>
|
||||||
|
private async Task<long> CurrentHeadAsync(Guid vaultId, CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
var head = await database.VaultChanges
|
||||||
.Where(c => c.VaultId == vaultId)
|
.Where(c => c.VaultId == vaultId)
|
||||||
.OrderByDescending(c => c.Sequence)
|
.OrderByDescending(c => c.Sequence)
|
||||||
.Select(c => c.Sequence)
|
.Select(c => c.Sequence)
|
||||||
.FirstOrDefaultAsync(cancellationToken)
|
.FirstOrDefaultAsync(cancellationToken)
|
||||||
.ConfigureAwait(false);
|
.ConfigureAwait(false);
|
||||||
|
|
||||||
|
return Math.Max(head, await HorizonAsync(vaultId, cancellationToken).ConfigureAwait(false));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>The vault's horizon as the database holds it now, not as the request loaded it.</summary>
|
||||||
|
private Task<long> HorizonAsync(Guid vaultId, CancellationToken cancellationToken) =>
|
||||||
|
database.Vaults
|
||||||
|
.Where(v => v.Id == vaultId)
|
||||||
|
.Select(v => v.SyncHorizon)
|
||||||
|
.SingleAsync(cancellationToken);
|
||||||
|
|
||||||
private static SyncPushResult Invalid(SyncPushOperation operation, string detail) =>
|
private static SyncPushResult Invalid(SyncPushOperation operation, string detail) =>
|
||||||
new(
|
new(
|
||||||
operation.OperationId,
|
operation.OperationId,
|
||||||
|
|||||||
@@ -0,0 +1,198 @@
|
|||||||
|
using DodoSSH.Api.Setup;
|
||||||
|
using DodoSSH.Domain;
|
||||||
|
using DodoSSH.Infrastructure;
|
||||||
|
using Microsoft.EntityFrameworkCore;
|
||||||
|
using Microsoft.Extensions.Options;
|
||||||
|
|
||||||
|
namespace DodoSSH.Api.Features.Sync;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Removes tombstones past their retention, and change-log entries a later one has superseded.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// ADR 0003 promised this and nothing did it: every deletion stayed a row for ever, and every edit stayed a
|
||||||
|
/// change-log entry for ever, so the log grew without bound and a client starting from nothing walked every
|
||||||
|
/// edit ever made to read the items that exist.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// <b>Superseded entries go at any time.</b> A pull returns entries after a cursor in order, so a client
|
||||||
|
/// before both an item's old entry and its new one reads the new one, and a client between them already has
|
||||||
|
/// the old one. Nothing reads history: a conflict resolves against the newest entry alone.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// <b>Tombstones go after the retention</b>, and taking them is the one change here a client can notice: a
|
||||||
|
/// machine that last synced before the deletion would never hear of it. So each vault records the highest
|
||||||
|
/// sequence collected from it — <see cref="Vault.SyncHorizon"/> — and a pull from a cursor below that is
|
||||||
|
/// refused. The client answers by starting over, and a pull from nothing removes what the server no longer
|
||||||
|
/// has; see the client's <c>SyncEngine</c>.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// Per vault under the same advisory lock a push takes, so a collection never interleaves with a write to
|
||||||
|
/// the vault — and the items are read again under it, so one restored since it was found is left alone.
|
||||||
|
/// </para>
|
||||||
|
/// </remarks>
|
||||||
|
internal sealed class TombstoneCollector(
|
||||||
|
DodoDbContext database,
|
||||||
|
IOptions<SyncOptions> syncOptions,
|
||||||
|
TimeProvider clock,
|
||||||
|
ILogger<TombstoneCollector> logger)
|
||||||
|
{
|
||||||
|
/// <summary>Collects what is due, once.</summary>
|
||||||
|
/// <returns>How many tombstones and superseded entries went.</returns>
|
||||||
|
internal async Task<int> RunOnceAsync(CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
var before = clock.GetUtcNow() - TimeSpan.FromDays(syncOptions.Value.TombstoneRetentionDays);
|
||||||
|
|
||||||
|
var collected = 0;
|
||||||
|
|
||||||
|
collected += await CollectAsync(db => db.Hosts, ChangeEntityType.SshHost, before, cancellationToken).ConfigureAwait(false);
|
||||||
|
collected += await CollectAsync(db => db.SshKeys, ChangeEntityType.SshKey, before, cancellationToken).ConfigureAwait(false);
|
||||||
|
collected += await CollectAsync(db => db.Credentials, ChangeEntityType.Credential, before, cancellationToken).ConfigureAwait(false);
|
||||||
|
collected += await CollectAsync(db => db.KnownHostKeys, ChangeEntityType.KnownHostKey, before, cancellationToken).ConfigureAwait(false);
|
||||||
|
collected += await CollectAsync(db => db.HostGroups, ChangeEntityType.HostGroup, before, cancellationToken).ConfigureAwait(false);
|
||||||
|
collected += await CollectAsync(db => db.Tags, ChangeEntityType.Tag, before, cancellationToken).ConfigureAwait(false);
|
||||||
|
collected += await CollectAsync(db => db.Snippets, ChangeEntityType.Snippet, before, cancellationToken).ConfigureAwait(false);
|
||||||
|
collected += await CollectAsync(db => db.ConnectionLog, ChangeEntityType.ConnectionLogEntry, before, cancellationToken).ConfigureAwait(false);
|
||||||
|
collected += await CollectAsync(db => db.ActivityLog, ChangeEntityType.ActivityLogEntry, before, cancellationToken).ConfigureAwait(false);
|
||||||
|
collected += await CollectAsync(db => db.ObjectStores, ChangeEntityType.ObjectStore, before, cancellationToken).ConfigureAwait(false);
|
||||||
|
collected += await CollectAsync(db => db.SpelunkKeys, ChangeEntityType.SpelunkKey, before, cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
|
collected += await CompactAsync(cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
|
SyncLog.Collected(logger, collected);
|
||||||
|
|
||||||
|
return collected;
|
||||||
|
}
|
||||||
|
|
||||||
|
private async Task<int> CollectAsync<T>(
|
||||||
|
Func<DodoDbContext, DbSet<T>> table,
|
||||||
|
ChangeEntityType type,
|
||||||
|
DateTimeOffset before,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
where T : class, IVaultItem
|
||||||
|
{
|
||||||
|
var due = await table(database)
|
||||||
|
.Where(item => item.DeletedAtUtc != null && item.DeletedAtUtc < before)
|
||||||
|
.Select(item => new { item.VaultId, item.Id })
|
||||||
|
.ToListAsync(cancellationToken)
|
||||||
|
.ConfigureAwait(false);
|
||||||
|
|
||||||
|
var collected = 0;
|
||||||
|
|
||||||
|
foreach (var vault in due.GroupBy(item => item.VaultId))
|
||||||
|
{
|
||||||
|
var ids = vault.Select(item => item.Id).ToArray();
|
||||||
|
|
||||||
|
collected += await database.ExecuteFreshAsync(
|
||||||
|
[],
|
||||||
|
() => CollectFromVaultAsync(table, type, vault.Key, ids, before, cancellationToken))
|
||||||
|
.ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
return collected;
|
||||||
|
}
|
||||||
|
|
||||||
|
private async Task<int> CollectFromVaultAsync<T>(
|
||||||
|
Func<DodoDbContext, DbSet<T>> table,
|
||||||
|
ChangeEntityType type,
|
||||||
|
Guid vaultId,
|
||||||
|
Guid[] ids,
|
||||||
|
DateTimeOffset before,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
where T : class, IVaultItem
|
||||||
|
{
|
||||||
|
var transaction = await database.Database.BeginTransactionAsync(cancellationToken).ConfigureAwait(false);
|
||||||
|
await using var _ = transaction.ConfigureAwait(false);
|
||||||
|
|
||||||
|
await database.Database
|
||||||
|
.ExecuteSqlAsync($"SELECT pg_advisory_xact_lock(hashtextextended({vaultId.ToString()}, 0))", cancellationToken)
|
||||||
|
.ConfigureAwait(false);
|
||||||
|
|
||||||
|
// Read again under the lock: an item restored since it was found is no longer due.
|
||||||
|
var still = await table(database)
|
||||||
|
.Where(item => item.VaultId == vaultId
|
||||||
|
&& ids.Contains(item.Id)
|
||||||
|
&& item.DeletedAtUtc != null
|
||||||
|
&& item.DeletedAtUtc < before)
|
||||||
|
.Select(item => new { item.Id, item.ChangeSequence })
|
||||||
|
.ToListAsync(cancellationToken)
|
||||||
|
.ConfigureAwait(false);
|
||||||
|
|
||||||
|
if (still.Count == 0)
|
||||||
|
{
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
var going = still.Select(item => item.Id).ToArray();
|
||||||
|
var horizon = still.Max(item => item.ChangeSequence);
|
||||||
|
|
||||||
|
await table(database)
|
||||||
|
.Where(item => item.VaultId == vaultId && going.Contains(item.Id))
|
||||||
|
.ExecuteDeleteAsync(cancellationToken)
|
||||||
|
.ConfigureAwait(false);
|
||||||
|
|
||||||
|
await database.VaultChanges
|
||||||
|
.Where(change => change.VaultId == vaultId && change.EntityType == type && going.Contains(change.EntityId))
|
||||||
|
.ExecuteDeleteAsync(cancellationToken)
|
||||||
|
.ConfigureAwait(false);
|
||||||
|
|
||||||
|
await database.Vaults
|
||||||
|
.Where(v => v.Id == vaultId && v.SyncHorizon < horizon)
|
||||||
|
.ExecuteUpdateAsync(set => set.SetProperty(v => v.SyncHorizon, horizon), cancellationToken)
|
||||||
|
.ConfigureAwait(false);
|
||||||
|
|
||||||
|
await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
|
return still.Count;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Removes every change-log entry a later entry for the same item has superseded.</summary>
|
||||||
|
private Task<int> CompactAsync(CancellationToken cancellationToken) =>
|
||||||
|
database.Database.ExecuteSqlRawAsync(
|
||||||
|
"DELETE FROM " + DodoDbContext.SchemaName + ".sync_change AS older "
|
||||||
|
+ "USING " + DodoDbContext.SchemaName + ".sync_change AS newer "
|
||||||
|
+ "WHERE newer.vault_id = older.vault_id "
|
||||||
|
+ "AND newer.entity_type = older.entity_type "
|
||||||
|
+ "AND newer.entity_id = older.entity_id "
|
||||||
|
+ "AND newer.sequence > older.sequence",
|
||||||
|
cancellationToken);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Runs the <see cref="TombstoneCollector"/> on a schedule, in a scope of its own each time.</summary>
|
||||||
|
internal sealed class TombstoneCollectionService(
|
||||||
|
IServiceScopeFactory scopes,
|
||||||
|
IOptions<SyncOptions> syncOptions,
|
||||||
|
TimeProvider clock,
|
||||||
|
ILogger<TombstoneCollectionService> logger) : BackgroundService
|
||||||
|
{
|
||||||
|
/// <inheritdoc />
|
||||||
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||||
|
{
|
||||||
|
var interval = TimeSpan.FromMinutes(syncOptions.Value.TombstoneCollectionIntervalMinutes);
|
||||||
|
|
||||||
|
while (!stoppingToken.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await Task.Delay(interval, clock, stoppingToken).ConfigureAwait(false);
|
||||||
|
|
||||||
|
var scope = scopes.CreateAsyncScope();
|
||||||
|
await using var _ = scope.ConfigureAwait(false);
|
||||||
|
|
||||||
|
await scope.ServiceProvider.GetRequiredService<TombstoneCollector>()
|
||||||
|
.RunOnceAsync(stoppingToken)
|
||||||
|
.ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
catch (Exception exception) when (exception is not OutOfMemoryException)
|
||||||
|
{
|
||||||
|
// The next run tries again; nothing collected late is harmed by waiting.
|
||||||
|
SyncLog.CollectionFailed(logger, exception);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -48,6 +48,10 @@ builder.Services.AddSingleton<ICursorKeyProvider, CursorKeyProvider>();
|
|||||||
builder.Services.AddSingleton<VaultEventHub>();
|
builder.Services.AddSingleton<VaultEventHub>();
|
||||||
builder.Services.AddSingleton<IVaultEventPublisher, VaultEventFanOut>();
|
builder.Services.AddSingleton<IVaultEventPublisher, VaultEventFanOut>();
|
||||||
|
|
||||||
|
// ADR 0003's tombstone collection, and the compaction of the change log that goes with it.
|
||||||
|
builder.Services.AddScoped<TombstoneCollector>();
|
||||||
|
builder.Services.AddHostedService<TombstoneCollectionService>();
|
||||||
|
|
||||||
// The relay's services, its ticket key ring, and a shutdown long enough for its drain. See ADR 0004.
|
// The relay's services, its ticket key ring, and a shutdown long enough for its drain. See ADR 0004.
|
||||||
builder.Services.AddDodoRelay();
|
builder.Services.AddDodoRelay();
|
||||||
|
|
||||||
|
|||||||
@@ -289,6 +289,10 @@ public sealed class SyncOptions
|
|||||||
[Range(1, 3650)]
|
[Range(1, 3650)]
|
||||||
public int TombstoneRetentionDays { get; set; } = 90;
|
public int TombstoneRetentionDays { get; set; } = 90;
|
||||||
|
|
||||||
|
/// <summary>How often the tombstone collector runs, in minutes.</summary>
|
||||||
|
[Range(1, 7 * 24 * 60)]
|
||||||
|
public int TombstoneCollectionIntervalMinutes { get; set; } = 6 * 60;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Key used to sign sync cursors, base64.
|
/// Key used to sign sync cursors, base64.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
|||||||
@@ -6451,6 +6451,8 @@ internal sealed partial class VaultViewModel(
|
|||||||
|| pass.PushedItems > 0
|
|| pass.PushedItems > 0
|
||||||
|| pass.Merged > 0
|
|| pass.Merged > 0
|
||||||
|| pass.Resurrected > 0
|
|| pass.Resurrected > 0
|
||||||
|
|| pass.Recreated > 0
|
||||||
|
|| pass.Pruned > 0
|
||||||
|| pass.DeletesAbandoned > 0
|
|| pass.DeletesAbandoned > 0
|
||||||
|| pass.Parked > 0
|
|| pass.Parked > 0
|
||||||
|| pass.ResyncedFromStart);
|
|| pass.ResyncedFromStart);
|
||||||
@@ -13159,13 +13161,27 @@ internal sealed partial class VaultViewModel(
|
|||||||
+ "the beginning. "
|
+ "the beginning. "
|
||||||
: string.Empty;
|
: string.Empty;
|
||||||
|
|
||||||
|
// Not a matter for attention — the server had already deleted these, and the machine is only catching
|
||||||
|
// up — but said, because items going from the list unannounced looks like losing them.
|
||||||
|
var pruned = report.Pruned > 0
|
||||||
|
? $" {report.Pruned} item(s) deleted elsewhere while this machine was away were removed."
|
||||||
|
: string.Empty;
|
||||||
|
|
||||||
if (!report.NeedsAttention)
|
if (!report.NeedsAttention)
|
||||||
{
|
{
|
||||||
return replayed + (report.Pulled == 0 && report.Pushed == 0
|
return replayed + (report.Pulled == 0 && report.Pushed == 0 && report.Pruned == 0
|
||||||
? "Already up to date."
|
? "Already up to date."
|
||||||
: $"Synchronised: {report.Pulled} in, {report.Pushed} out.");
|
: $"Synchronised: {report.Pulled} in, {report.Pushed} out.") + pruned;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var notes = AttentionNotes(report);
|
||||||
|
|
||||||
|
return replayed + "Synchronised, but: " + string.Join("; ", notes) + "." + pruned;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>What a pass that needs attention says, one clause per thing.</summary>
|
||||||
|
private static List<string> AttentionNotes(SyncReport report)
|
||||||
|
{
|
||||||
var notes = new List<string>();
|
var notes = new List<string>();
|
||||||
|
|
||||||
// "item(s)", not "host(s)": a vault now holds keys as well, and a report that named the wrong kind
|
// "item(s)", not "host(s)": a vault now holds keys as well, and a report that named the wrong kind
|
||||||
@@ -13175,6 +13191,11 @@ internal sealed partial class VaultViewModel(
|
|||||||
notes.Add($"{report.Resurrected} item(s) deleted elsewhere were kept under a new name");
|
notes.Add($"{report.Resurrected} item(s) deleted elsewhere were kept under a new name");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (report.Recreated > 0)
|
||||||
|
{
|
||||||
|
notes.Add($"{report.Recreated} item(s) deleted elsewhere long ago were put back, with this machine's changes");
|
||||||
|
}
|
||||||
|
|
||||||
if (report.DeletesAbandoned > 0)
|
if (report.DeletesAbandoned > 0)
|
||||||
{
|
{
|
||||||
notes.Add($"{report.DeletesAbandoned} deletion(s) were not applied because of a newer edit");
|
notes.Add($"{report.DeletesAbandoned} deletion(s) were not applied because of a newer edit");
|
||||||
@@ -13200,7 +13221,7 @@ internal sealed partial class VaultViewModel(
|
|||||||
+ "before you can see anything written since");
|
+ "before you can see anything written since");
|
||||||
}
|
}
|
||||||
|
|
||||||
return replayed + "Synchronised, but: " + string.Join("; ", notes) + ".";
|
return notes;
|
||||||
}
|
}
|
||||||
|
|
||||||
private async Task RunAsync(string busyMessage, Func<Task> work)
|
private async Task RunAsync(string busyMessage, Func<Task> work)
|
||||||
|
|||||||
@@ -240,6 +240,15 @@ internal sealed class CachedItemRow
|
|||||||
public bool IsDeleted { get; set; }
|
public bool IsDeleted { get; set; }
|
||||||
|
|
||||||
public DateTimeOffset UpdatedAtUtc { get; set; }
|
public DateTimeOffset UpdatedAtUtc { get; set; }
|
||||||
|
|
||||||
|
/// <summary>The last full pull that saw this item on the server; null if none has since this was kept.</summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// The server collects tombstones after their retention (ADR 0003), so a machine away longer than that
|
||||||
|
/// is never told of some deletions: it is refused, starts over, and pulls everything that still exists.
|
||||||
|
/// What the pull from nothing did not mark is what the server no longer has. See
|
||||||
|
/// <c>ItemStore.SweepAsync</c>.
|
||||||
|
/// </remarks>
|
||||||
|
public long? SweepMark { get; set; }
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
@@ -361,6 +370,20 @@ internal sealed class SyncStateRow
|
|||||||
/// beside the cursor so a change of list starts the pull over; see <c>SyncEngine.PullAsync</c>.
|
/// beside the cursor so a change of list starts the pull over; see <c>SyncEngine.PullAsync</c>.
|
||||||
/// </remarks>
|
/// </remarks>
|
||||||
public string? PulledTypes { get; set; }
|
public string? PulledTypes { get; set; }
|
||||||
|
|
||||||
|
/// <summary>The mark of the pull from nothing in progress, or null when none is.</summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// Kept here rather than in memory because a pull from nothing can take several passes — the pages per
|
||||||
|
/// pass are capped — and what it has seen has to survive between them. See <see cref="CachedItemRow.SweepMark"/>.
|
||||||
|
/// </remarks>
|
||||||
|
public long? SweepMark { get; set; }
|
||||||
|
|
||||||
|
/// <summary>The highest change sequence that pull has read so far.</summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// The sweep's upper bound: an item written here later than this — one this machine pushed while the
|
||||||
|
/// pull was under way — is newer than anything the pull has read, and not evidence of anything.
|
||||||
|
/// </remarks>
|
||||||
|
public long SweepSeenUpTo { get; set; }
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>Something the merge had to override, or an item that could not be processed.</summary>
|
/// <summary>Something the merge had to override, or an item that could not be processed.</summary>
|
||||||
|
|||||||
@@ -177,6 +177,94 @@ public sealed class ItemStore(
|
|||||||
await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
|
await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>Marks the items a pull from nothing has just read as ones the server still has.</summary>
|
||||||
|
/// <remarks>One update per type in the page; see <see cref="SweepAsync"/>.</remarks>
|
||||||
|
public async Task MarkSeenAsync(
|
||||||
|
Guid vaultId,
|
||||||
|
long mark,
|
||||||
|
IReadOnlyCollection<(SyncEntityType Type, Guid Id)> seen,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(seen);
|
||||||
|
|
||||||
|
if (seen.Count == 0)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
var context = contexts.CreateDbContext();
|
||||||
|
await using var scope = context.ConfigureAwait(false);
|
||||||
|
|
||||||
|
foreach (var group in seen.GroupBy(item => item.Type))
|
||||||
|
{
|
||||||
|
var type = group.Key;
|
||||||
|
var ids = group.Select(item => item.Id).Distinct().ToList();
|
||||||
|
|
||||||
|
await context.Set<CachedItemRow>()
|
||||||
|
.Where(row => row.VaultId == vaultId && row.EntityType == type && ids.Contains(row.EntityId))
|
||||||
|
.ExecuteUpdateAsync(set => set.SetProperty(row => row.SweepMark, mark), cancellationToken)
|
||||||
|
.ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>The highest change sequence held for a vault, or zero for none.</summary>
|
||||||
|
/// <remarks>Where a pull from nothing starts its sweep's upper bound; see <see cref="SweepAsync"/>.</remarks>
|
||||||
|
public async Task<long> HighestSequenceAsync(Guid vaultId, CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
var context = contexts.CreateDbContext();
|
||||||
|
await using var scope = context.ConfigureAwait(false);
|
||||||
|
|
||||||
|
return await context.Set<CachedItemRow>()
|
||||||
|
.Where(row => row.VaultId == vaultId)
|
||||||
|
.MaxAsync(row => (long?)row.ChangeSequence, cancellationToken)
|
||||||
|
.ConfigureAwait(false) ?? 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Removes the items a completed pull from nothing did not see, which are the ones the server no longer has.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// The other half of the server's tombstone collection. A machine that missed a deletion for longer than
|
||||||
|
/// the retention is refused its cursor and pulls everything again — and everything again does not include
|
||||||
|
/// the item, so its copy here would otherwise last for ever, and its next edit would push it back.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// â—† Two kinds of row are not evidence and are kept. One newer than <paramref name="seenUpTo"/> was
|
||||||
|
/// written by this machine's own push during the pull: the bound is the higher of what the cache held
|
||||||
|
/// when the pull began and what it has read since, and the server's head is past both, so a push lands
|
||||||
|
/// above it. One with a change still waiting to go is the user's work: it is
|
||||||
|
/// left for the push, which finds the server without the item and says so — see
|
||||||
|
/// <c>SyncEngine.RetryAsCreateAsync</c> — rather than being thrown away here unseen.
|
||||||
|
/// </para>
|
||||||
|
/// </remarks>
|
||||||
|
/// <returns>How many items went.</returns>
|
||||||
|
public async Task<int> SweepAsync(
|
||||||
|
Guid vaultId,
|
||||||
|
IReadOnlyCollection<SyncEntityType> types,
|
||||||
|
long mark,
|
||||||
|
long seenUpTo,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(types);
|
||||||
|
|
||||||
|
var context = contexts.CreateDbContext();
|
||||||
|
await using var scope = context.ConfigureAwait(false);
|
||||||
|
|
||||||
|
// Only the types the pull asked for: a row of any other type was never going to be marked.
|
||||||
|
return await context.Set<CachedItemRow>()
|
||||||
|
.Where(row => row.VaultId == vaultId
|
||||||
|
&& types.Contains(row.EntityType)
|
||||||
|
&& (row.SweepMark == null || row.SweepMark != mark)
|
||||||
|
&& row.ChangeSequence <= seenUpTo
|
||||||
|
&& !context.Set<OutboxRow>().Any(pending =>
|
||||||
|
pending.VaultId == row.VaultId
|
||||||
|
&& pending.EntityType == row.EntityType
|
||||||
|
&& pending.EntityId == row.EntityId))
|
||||||
|
.ExecuteDeleteAsync(cancellationToken)
|
||||||
|
.ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>Removes a tombstone whose change has been seen by everything that needed it.</summary>
|
/// <summary>Removes a tombstone whose change has been seen by everything that needed it.</summary>
|
||||||
/// <remarks>
|
/// <remarks>
|
||||||
/// Only ever called for a row that is already a tombstone. Collecting a live item here would make
|
/// Only ever called for a row that is already a tombstone. Collecting a live item here would make
|
||||||
|
|||||||
+481
@@ -0,0 +1,481 @@
|
|||||||
|
// <auto-generated />
|
||||||
|
using System;
|
||||||
|
using DodoSSH.Client.Storage;
|
||||||
|
using Microsoft.EntityFrameworkCore;
|
||||||
|
using Microsoft.EntityFrameworkCore.Infrastructure;
|
||||||
|
using Microsoft.EntityFrameworkCore.Migrations;
|
||||||
|
using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
|
||||||
|
|
||||||
|
#nullable disable
|
||||||
|
|
||||||
|
namespace DodoSSH.Client.Storage.Migrations
|
||||||
|
{
|
||||||
|
[DbContext(typeof(ClientCacheContext))]
|
||||||
|
[Migration("20261002152108_AddSweepMarks")]
|
||||||
|
partial class AddSweepMarks
|
||||||
|
{
|
||||||
|
/// <inheritdoc />
|
||||||
|
protected override void BuildTargetModel(ModelBuilder modelBuilder)
|
||||||
|
{
|
||||||
|
#pragma warning disable 612, 618
|
||||||
|
modelBuilder.HasAnnotation("ProductVersion", "10.0.10");
|
||||||
|
|
||||||
|
modelBuilder.Entity("DodoSSH.Client.Storage.CachedItemRow", b =>
|
||||||
|
{
|
||||||
|
b.Property<Guid>("VaultId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("vault_id");
|
||||||
|
|
||||||
|
b.Property<int>("EntityType")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("entity_type");
|
||||||
|
|
||||||
|
b.Property<Guid>("EntityId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("entity_id");
|
||||||
|
|
||||||
|
b.Property<byte>("AadVersion")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("aad_version");
|
||||||
|
|
||||||
|
b.Property<long>("ChangeSequence")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("change_sequence");
|
||||||
|
|
||||||
|
b.Property<Guid?>("DataKeyId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("data_key_id");
|
||||||
|
|
||||||
|
b.Property<bool>("IsDeleted")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("is_deleted");
|
||||||
|
|
||||||
|
b.Property<uint>("KeyGeneration")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("key_generation");
|
||||||
|
|
||||||
|
b.Property<byte[]>("Payload")
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("payload");
|
||||||
|
|
||||||
|
b.Property<byte[]>("ProtectedFields")
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("protected_fields");
|
||||||
|
|
||||||
|
b.Property<long?>("SweepMark")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("sweep_mark");
|
||||||
|
|
||||||
|
b.Property<long>("UpdatedAtUtc")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("updated_at_utc");
|
||||||
|
|
||||||
|
b.Property<int>("Version")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("version");
|
||||||
|
|
||||||
|
b.Property<byte[]>("WrappedDataKey")
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("wrapped_data_key");
|
||||||
|
|
||||||
|
b.HasKey("VaultId", "EntityType", "EntityId")
|
||||||
|
.HasName("pk_item");
|
||||||
|
|
||||||
|
b.HasIndex("VaultId", "ChangeSequence")
|
||||||
|
.HasDatabaseName("ix_item_vault_id_change_sequence");
|
||||||
|
|
||||||
|
b.HasIndex("VaultId", "EntityType")
|
||||||
|
.HasDatabaseName("ix_item_vault_id_entity_type");
|
||||||
|
|
||||||
|
b.ToTable("item", (string)null);
|
||||||
|
});
|
||||||
|
|
||||||
|
modelBuilder.Entity("DodoSSH.Client.Storage.CachedVaultKeyWrapRow", b =>
|
||||||
|
{
|
||||||
|
b.Property<Guid>("VaultId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("vault_id");
|
||||||
|
|
||||||
|
b.Property<uint>("KeyGeneration")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("key_generation");
|
||||||
|
|
||||||
|
b.Property<byte[]>("WrappedKey")
|
||||||
|
.IsRequired()
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("wrapped_key");
|
||||||
|
|
||||||
|
b.HasKey("VaultId", "KeyGeneration")
|
||||||
|
.HasName("pk_vault_key_wrap");
|
||||||
|
|
||||||
|
b.ToTable("vault_key_wrap", (string)null);
|
||||||
|
});
|
||||||
|
|
||||||
|
modelBuilder.Entity("DodoSSH.Client.Storage.CachedVaultRow", b =>
|
||||||
|
{
|
||||||
|
b.Property<Guid>("VaultId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("vault_id");
|
||||||
|
|
||||||
|
b.Property<bool>("Hidden")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("hidden");
|
||||||
|
|
||||||
|
b.Property<bool>("IsPersonal")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("is_personal");
|
||||||
|
|
||||||
|
b.Property<uint>("KeyGeneration")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("key_generation");
|
||||||
|
|
||||||
|
b.Property<string>("Name")
|
||||||
|
.IsRequired()
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("name");
|
||||||
|
|
||||||
|
b.Property<int>("Permissions")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("permissions");
|
||||||
|
|
||||||
|
b.Property<bool>("RekeyRequired")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("rekey_required");
|
||||||
|
|
||||||
|
b.Property<Guid?>("TeamId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("team_id");
|
||||||
|
|
||||||
|
b.Property<long>("UpdatedAtUtc")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("updated_at_utc");
|
||||||
|
|
||||||
|
b.Property<byte[]>("WrappedVaultKey")
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("wrapped_vault_key");
|
||||||
|
|
||||||
|
b.HasKey("VaultId")
|
||||||
|
.HasName("pk_vault");
|
||||||
|
|
||||||
|
b.ToTable("vault", (string)null);
|
||||||
|
});
|
||||||
|
|
||||||
|
modelBuilder.Entity("DodoSSH.Client.Storage.ConflictRow", b =>
|
||||||
|
{
|
||||||
|
b.Property<Guid>("Id")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("id");
|
||||||
|
|
||||||
|
b.Property<bool>("Acknowledged")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("acknowledged");
|
||||||
|
|
||||||
|
b.Property<byte[]>("Detail")
|
||||||
|
.IsRequired()
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("detail");
|
||||||
|
|
||||||
|
b.Property<long>("DetectedAtUtc")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("detected_at_utc");
|
||||||
|
|
||||||
|
b.Property<Guid>("EntityId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("entity_id");
|
||||||
|
|
||||||
|
b.Property<int>("EntityType")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("entity_type");
|
||||||
|
|
||||||
|
b.Property<int>("Kind")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("kind");
|
||||||
|
|
||||||
|
b.Property<Guid>("VaultId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("vault_id");
|
||||||
|
|
||||||
|
b.HasKey("Id")
|
||||||
|
.HasName("pk_conflict");
|
||||||
|
|
||||||
|
b.HasIndex("VaultId", "Acknowledged")
|
||||||
|
.HasDatabaseName("ix_conflict_vault_id_acknowledged");
|
||||||
|
|
||||||
|
b.HasIndex("VaultId", "EntityType", "EntityId")
|
||||||
|
.HasDatabaseName("ix_conflict_vault_id_entity_type_entity_id");
|
||||||
|
|
||||||
|
b.ToTable("conflict", (string)null);
|
||||||
|
});
|
||||||
|
|
||||||
|
modelBuilder.Entity("DodoSSH.Client.Storage.OutboxRow", b =>
|
||||||
|
{
|
||||||
|
b.Property<long>("Sequence")
|
||||||
|
.ValueGeneratedOnAdd()
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("sequence");
|
||||||
|
|
||||||
|
b.Property<byte>("AadVersion")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("aad_version");
|
||||||
|
|
||||||
|
b.Property<byte?>("AncestorAadVersion")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("ancestor_aad_version");
|
||||||
|
|
||||||
|
b.Property<Guid?>("AncestorDataKeyId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("ancestor_data_key_id");
|
||||||
|
|
||||||
|
b.Property<uint?>("AncestorKeyGeneration")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("ancestor_key_generation");
|
||||||
|
|
||||||
|
b.Property<byte[]>("AncestorPayload")
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("ancestor_payload");
|
||||||
|
|
||||||
|
b.Property<byte[]>("AncestorProtectedFields")
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("ancestor_protected_fields");
|
||||||
|
|
||||||
|
b.Property<int?>("AncestorVersion")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("ancestor_version");
|
||||||
|
|
||||||
|
b.Property<byte[]>("AncestorWrappedDataKey")
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("ancestor_wrapped_data_key");
|
||||||
|
|
||||||
|
b.Property<int>("Attempts")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("attempts");
|
||||||
|
|
||||||
|
b.Property<Guid?>("DataKeyId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("data_key_id");
|
||||||
|
|
||||||
|
b.Property<Guid>("EntityId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("entity_id");
|
||||||
|
|
||||||
|
b.Property<int>("EntityType")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("entity_type");
|
||||||
|
|
||||||
|
b.Property<int?>("ExpectedVersion")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("expected_version");
|
||||||
|
|
||||||
|
b.Property<bool>("IsParked")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("is_parked");
|
||||||
|
|
||||||
|
b.Property<uint>("KeyGeneration")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("key_generation");
|
||||||
|
|
||||||
|
b.Property<string>("LastError")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("last_error");
|
||||||
|
|
||||||
|
b.Property<int>("Operation")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("operation");
|
||||||
|
|
||||||
|
b.Property<Guid>("OperationId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("operation_id");
|
||||||
|
|
||||||
|
b.Property<byte[]>("Payload")
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("payload");
|
||||||
|
|
||||||
|
b.Property<byte[]>("ProtectedFields")
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("protected_fields");
|
||||||
|
|
||||||
|
b.Property<long>("QueuedAtUtc")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("queued_at_utc");
|
||||||
|
|
||||||
|
b.Property<Guid>("VaultId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("vault_id");
|
||||||
|
|
||||||
|
b.Property<byte[]>("WrappedDataKey")
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("wrapped_data_key");
|
||||||
|
|
||||||
|
b.HasKey("Sequence")
|
||||||
|
.HasName("pk_outbox");
|
||||||
|
|
||||||
|
b.HasIndex("OperationId")
|
||||||
|
.IsUnique()
|
||||||
|
.HasDatabaseName("ix_outbox_operation_id");
|
||||||
|
|
||||||
|
b.HasIndex("VaultId", "EntityType", "EntityId")
|
||||||
|
.IsUnique()
|
||||||
|
.HasDatabaseName("ix_outbox_vault_id_entity_type_entity_id");
|
||||||
|
|
||||||
|
b.HasIndex("VaultId", "IsParked", "Sequence")
|
||||||
|
.HasDatabaseName("ix_outbox_vault_id_is_parked_sequence");
|
||||||
|
|
||||||
|
b.ToTable("outbox", (string)null);
|
||||||
|
});
|
||||||
|
|
||||||
|
modelBuilder.Entity("DodoSSH.Client.Storage.RememberedSignInRow", b =>
|
||||||
|
{
|
||||||
|
b.Property<int>("Id")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("id");
|
||||||
|
|
||||||
|
b.Property<byte[]>("SealedRefreshToken")
|
||||||
|
.IsRequired()
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("sealed_refresh_token");
|
||||||
|
|
||||||
|
b.Property<long>("UpdatedAtUtc")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("updated_at_utc");
|
||||||
|
|
||||||
|
b.HasKey("Id")
|
||||||
|
.HasName("pk_remembered_sign_in");
|
||||||
|
|
||||||
|
b.ToTable("remembered_sign_in", null, t =>
|
||||||
|
{
|
||||||
|
t.HasCheckConstraint("ck_remembered_sign_in_singleton", "id = 1");
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
modelBuilder.Entity("DodoSSH.Client.Storage.SyncStateRow", b =>
|
||||||
|
{
|
||||||
|
b.Property<Guid>("VaultId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("vault_id");
|
||||||
|
|
||||||
|
b.Property<string>("Cursor")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("cursor");
|
||||||
|
|
||||||
|
b.Property<uint>("KeyGeneration")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("key_generation");
|
||||||
|
|
||||||
|
b.Property<long?>("LastPulledAtUtc")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("last_pulled_at_utc");
|
||||||
|
|
||||||
|
b.Property<long?>("LastPushedAtUtc")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("last_pushed_at_utc");
|
||||||
|
|
||||||
|
b.Property<string>("PulledTypes")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("pulled_types");
|
||||||
|
|
||||||
|
b.Property<long>("ServerTimeSkewMs")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("server_time_skew_ms");
|
||||||
|
|
||||||
|
b.Property<long?>("SweepMark")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("sweep_mark");
|
||||||
|
|
||||||
|
b.Property<long>("SweepSeenUpTo")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("sweep_seen_up_to");
|
||||||
|
|
||||||
|
b.HasKey("VaultId")
|
||||||
|
.HasName("pk_sync_state");
|
||||||
|
|
||||||
|
b.ToTable("sync_state", (string)null);
|
||||||
|
});
|
||||||
|
|
||||||
|
modelBuilder.Entity("DodoSSH.Client.Storage.UnlockMaterialRow", b =>
|
||||||
|
{
|
||||||
|
b.Property<int>("Id")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("id");
|
||||||
|
|
||||||
|
b.Property<Guid?>("DeviceId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("device_id");
|
||||||
|
|
||||||
|
b.Property<byte[]>("DeviceWrappedPrivateKey")
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("device_wrapped_private_key");
|
||||||
|
|
||||||
|
b.Property<string>("DisplayName")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("display_name");
|
||||||
|
|
||||||
|
b.Property<string>("Email")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("email");
|
||||||
|
|
||||||
|
b.Property<string>("Issuer")
|
||||||
|
.IsRequired()
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("issuer");
|
||||||
|
|
||||||
|
b.Property<string>("KdfAlgorithm")
|
||||||
|
.IsRequired()
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("kdf_algorithm");
|
||||||
|
|
||||||
|
b.Property<int>("KdfMemoryKibibytes")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("kdf_memory_kibibytes");
|
||||||
|
|
||||||
|
b.Property<int>("KdfParallelism")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("kdf_parallelism");
|
||||||
|
|
||||||
|
b.Property<int>("KdfPasses")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("kdf_passes");
|
||||||
|
|
||||||
|
b.Property<byte[]>("KdfSalt")
|
||||||
|
.IsRequired()
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("kdf_salt");
|
||||||
|
|
||||||
|
b.Property<uint>("KeyGeneration")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("key_generation");
|
||||||
|
|
||||||
|
b.Property<string>("ServerUrl")
|
||||||
|
.IsRequired()
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("server_url");
|
||||||
|
|
||||||
|
b.Property<string>("Subject")
|
||||||
|
.IsRequired()
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("subject");
|
||||||
|
|
||||||
|
b.Property<long>("UpdatedAtUtc")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("updated_at_utc");
|
||||||
|
|
||||||
|
b.Property<Guid>("UserId")
|
||||||
|
.HasColumnType("TEXT")
|
||||||
|
.HasColumnName("user_id");
|
||||||
|
|
||||||
|
b.Property<byte[]>("WrappedPrivateKey")
|
||||||
|
.IsRequired()
|
||||||
|
.HasColumnType("BLOB")
|
||||||
|
.HasColumnName("wrapped_private_key");
|
||||||
|
|
||||||
|
b.HasKey("Id")
|
||||||
|
.HasName("pk_unlock_material");
|
||||||
|
|
||||||
|
b.ToTable("unlock_material", null, t =>
|
||||||
|
{
|
||||||
|
t.HasCheckConstraint("ck_unlock_material_singleton", "id = 1");
|
||||||
|
});
|
||||||
|
});
|
||||||
|
#pragma warning restore 612, 618
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,49 @@
|
|||||||
|
using Microsoft.EntityFrameworkCore.Migrations;
|
||||||
|
|
||||||
|
#nullable disable
|
||||||
|
|
||||||
|
namespace DodoSSH.Client.Storage.Migrations
|
||||||
|
{
|
||||||
|
/// <inheritdoc />
|
||||||
|
public partial class AddSweepMarks : Migration
|
||||||
|
{
|
||||||
|
/// <inheritdoc />
|
||||||
|
protected override void Up(MigrationBuilder migrationBuilder)
|
||||||
|
{
|
||||||
|
migrationBuilder.AddColumn<long>(
|
||||||
|
name: "sweep_mark",
|
||||||
|
table: "sync_state",
|
||||||
|
type: "INTEGER",
|
||||||
|
nullable: true);
|
||||||
|
|
||||||
|
migrationBuilder.AddColumn<long>(
|
||||||
|
name: "sweep_seen_up_to",
|
||||||
|
table: "sync_state",
|
||||||
|
type: "INTEGER",
|
||||||
|
nullable: false,
|
||||||
|
defaultValue: 0L);
|
||||||
|
|
||||||
|
migrationBuilder.AddColumn<long>(
|
||||||
|
name: "sweep_mark",
|
||||||
|
table: "item",
|
||||||
|
type: "INTEGER",
|
||||||
|
nullable: true);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
protected override void Down(MigrationBuilder migrationBuilder)
|
||||||
|
{
|
||||||
|
migrationBuilder.DropColumn(
|
||||||
|
name: "sweep_mark",
|
||||||
|
table: "sync_state");
|
||||||
|
|
||||||
|
migrationBuilder.DropColumn(
|
||||||
|
name: "sweep_seen_up_to",
|
||||||
|
table: "sync_state");
|
||||||
|
|
||||||
|
migrationBuilder.DropColumn(
|
||||||
|
name: "sweep_mark",
|
||||||
|
table: "item");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -59,6 +59,10 @@ namespace DodoSSH.Client.Storage.Migrations
|
|||||||
.HasColumnType("BLOB")
|
.HasColumnType("BLOB")
|
||||||
.HasColumnName("protected_fields");
|
.HasColumnName("protected_fields");
|
||||||
|
|
||||||
|
b.Property<long?>("SweepMark")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("sweep_mark");
|
||||||
|
|
||||||
b.Property<long>("UpdatedAtUtc")
|
b.Property<long>("UpdatedAtUtc")
|
||||||
.HasColumnType("INTEGER")
|
.HasColumnType("INTEGER")
|
||||||
.HasColumnName("updated_at_utc");
|
.HasColumnName("updated_at_utc");
|
||||||
@@ -370,6 +374,14 @@ namespace DodoSSH.Client.Storage.Migrations
|
|||||||
.HasColumnType("INTEGER")
|
.HasColumnType("INTEGER")
|
||||||
.HasColumnName("server_time_skew_ms");
|
.HasColumnName("server_time_skew_ms");
|
||||||
|
|
||||||
|
b.Property<long?>("SweepMark")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("sweep_mark");
|
||||||
|
|
||||||
|
b.Property<long>("SweepSeenUpTo")
|
||||||
|
.HasColumnType("INTEGER")
|
||||||
|
.HasColumnName("sweep_seen_up_to");
|
||||||
|
|
||||||
b.HasKey("VaultId")
|
b.HasKey("VaultId")
|
||||||
.HasName("pk_sync_state");
|
.HasName("pk_sync_state");
|
||||||
|
|
||||||
|
|||||||
@@ -35,6 +35,12 @@ public enum ConflictKind
|
|||||||
|
|
||||||
/// <summary>The server refused the operation outright. Retrying will not help.</summary>
|
/// <summary>The server refused the operation outright. Retrying will not help.</summary>
|
||||||
Rejected = 6,
|
Rejected = 6,
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Someone else deleted an item this machine had edited, so long ago that the server had already let go
|
||||||
|
/// of the deletion. The edit was sent as the item again, under its own name and id.
|
||||||
|
/// </summary>
|
||||||
|
RecreatedAfterCollection = 7,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>What an offline unlock needs.</summary>
|
/// <summary>What an offline unlock needs.</summary>
|
||||||
@@ -191,7 +197,9 @@ public sealed record StoredSyncState(
|
|||||||
DateTimeOffset? LastPulledAt = null,
|
DateTimeOffset? LastPulledAt = null,
|
||||||
DateTimeOffset? LastPushedAt = null,
|
DateTimeOffset? LastPushedAt = null,
|
||||||
long ServerTimeSkewMs = 0,
|
long ServerTimeSkewMs = 0,
|
||||||
string? PulledTypes = null);
|
string? PulledTypes = null,
|
||||||
|
long? SweepMark = null,
|
||||||
|
long SweepSeenUpTo = 0);
|
||||||
|
|
||||||
/// <summary>Something the merge overrode, or an item that could not be processed.</summary>
|
/// <summary>Something the merge overrode, or an item that could not be processed.</summary>
|
||||||
/// <param name="Id">This record's own id, which its sealed detail is bound to.</param>
|
/// <param name="Id">This record's own id, which its sealed detail is bound to.</param>
|
||||||
|
|||||||
@@ -47,7 +47,9 @@ public sealed class SyncStateStore(IDbContextFactory<ClientCacheContext> context
|
|||||||
row.LastPulledAtUtc,
|
row.LastPulledAtUtc,
|
||||||
row.LastPushedAtUtc,
|
row.LastPushedAtUtc,
|
||||||
row.ServerTimeSkewMs,
|
row.ServerTimeSkewMs,
|
||||||
row.PulledTypes);
|
row.PulledTypes,
|
||||||
|
row.SweepMark,
|
||||||
|
row.SweepSeenUpTo);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>Records a vault's position.</summary>
|
/// <summary>Records a vault's position.</summary>
|
||||||
@@ -74,6 +76,8 @@ public sealed class SyncStateStore(IDbContextFactory<ClientCacheContext> context
|
|||||||
row.LastPushedAtUtc = state.LastPushedAt;
|
row.LastPushedAtUtc = state.LastPushedAt;
|
||||||
row.ServerTimeSkewMs = state.ServerTimeSkewMs;
|
row.ServerTimeSkewMs = state.ServerTimeSkewMs;
|
||||||
row.PulledTypes = state.PulledTypes;
|
row.PulledTypes = state.PulledTypes;
|
||||||
|
row.SweepMark = state.SweepMark;
|
||||||
|
row.SweepSeenUpTo = state.SweepSeenUpTo;
|
||||||
|
|
||||||
await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
|
await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
using System.Globalization;
|
using System.Globalization;
|
||||||
using System.Runtime.InteropServices;
|
using System.Runtime.InteropServices;
|
||||||
|
using System.Security.Cryptography;
|
||||||
using DodoSSH.Client.Api;
|
using DodoSSH.Client.Api;
|
||||||
using DodoSSH.Client.Storage;
|
using DodoSSH.Client.Storage;
|
||||||
using DodoSSH.Contracts;
|
using DodoSSH.Contracts;
|
||||||
@@ -155,23 +156,12 @@ public sealed class SyncEngine
|
|||||||
}
|
}
|
||||||
|
|
||||||
await ApplyPageAsync(vaultId, response.Changes, report, cancellationToken).ConfigureAwait(false);
|
await ApplyPageAsync(vaultId, response.Changes, report, cancellationToken).ConfigureAwait(false);
|
||||||
|
CountPulled(response.Changes, report);
|
||||||
foreach (var change in response.Changes)
|
|
||||||
{
|
|
||||||
report.Pulled++;
|
|
||||||
|
|
||||||
// Counted apart for the same reason the pushed ones are: a machine reads back the log
|
|
||||||
// entries it just wrote, so a pass that "pulled five changes" may have carried nothing
|
|
||||||
// anybody did. See SyncReport.PulledItems.
|
|
||||||
if (IsLog(change.EntityType))
|
|
||||||
{
|
|
||||||
report.PulledLogEntries++;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
var advanced = !string.Equals(state.Cursor, response.NextCursor, StringComparison.Ordinal);
|
var advanced = !string.Equals(state.Cursor, response.NextCursor, StringComparison.Ordinal);
|
||||||
|
|
||||||
state = Record(vaultId, state, response, report);
|
state = Record(vaultId, state, response, report);
|
||||||
|
state = await CarrySweepAsync(vaultId, state, response, report, cancellationToken).ConfigureAwait(false);
|
||||||
await syncState.SaveAsync(state, cancellationToken).ConfigureAwait(false);
|
await syncState.SaveAsync(state, cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
if (!response.HasMore)
|
if (!response.HasMore)
|
||||||
@@ -188,15 +178,22 @@ public sealed class SyncEngine
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
private static void CountPulled(IReadOnlyList<SyncChange> changes, SyncReportBuilder report)
|
||||||
/// Whether the server refused the cursor this client sent, rather than failing for some other reason.
|
{
|
||||||
/// </summary>
|
foreach (var change in changes)
|
||||||
/// <remarks>
|
{
|
||||||
/// Read from the problem code and not from the prose, which is free to change, and never claimed for a
|
report.Pulled++;
|
||||||
/// request that carried no cursor: "from the beginning" is the one position a client is allowed to ask
|
|
||||||
/// for, so a rejection of <em>that</em> is a server this code cannot reason about and has to surface.
|
// Counted apart for the same reason the pushed ones are: a machine reads back the log entries it
|
||||||
/// It is also what keeps the retry from looping — the restarted request sends no cursor.
|
// just wrote, so a pass that "pulled five changes" may have carried nothing anybody did. See
|
||||||
/// </remarks>
|
// SyncReport.PulledItems.
|
||||||
|
if (IsLog(change.EntityType))
|
||||||
|
{
|
||||||
|
report.PulledLogEntries++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>Whether a change is one the machine wrote about itself rather than one somebody made.</summary>
|
/// <summary>Whether a change is one the machine wrote about itself rather than one somebody made.</summary>
|
||||||
/// <remarks>
|
/// <remarks>
|
||||||
/// Used only for reporting. Log entries sync exactly like every other item — they are pulled, pushed,
|
/// Used only for reporting. Log entries sync exactly like every other item — they are pulled, pushed,
|
||||||
@@ -206,10 +203,46 @@ public sealed class SyncEngine
|
|||||||
private static bool IsLog(SyncEntityType type) =>
|
private static bool IsLog(SyncEntityType type) =>
|
||||||
type is SyncEntityType.ConnectionLogEntry or SyncEntityType.ActivityLogEntry;
|
type is SyncEntityType.ConnectionLogEntry or SyncEntityType.ActivityLogEntry;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Whether the server refused the cursor this client sent, rather than failing for some other reason.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// Read from the problem code and not from the prose, which is free to change, and never claimed for a
|
||||||
|
/// request that carried no cursor: "from the beginning" is the one position a client is allowed to ask
|
||||||
|
/// for, so a rejection of <em>that</em> is a server this code cannot reason about and has to surface.
|
||||||
|
/// It is also what keeps the retry from looping — the restarted request sends no cursor.
|
||||||
|
/// </remarks>
|
||||||
private static bool WasTheCursorRefused(DodoSshApiException exception, string? cursor) =>
|
private static bool WasTheCursorRefused(DodoSshApiException exception, string? cursor) =>
|
||||||
!string.IsNullOrEmpty(cursor)
|
!string.IsNullOrEmpty(cursor)
|
||||||
&& string.Equals(exception.Code, ProblemCodes.InvalidCursor, StringComparison.Ordinal);
|
&& string.Equals(exception.Code, ProblemCodes.InvalidCursor, StringComparison.Ordinal);
|
||||||
|
|
||||||
|
/// <summary>Where the pull should carry on from, which is the start when the types it reads have changed.</summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// A cursor read under a different list of types is past changes of the types it did not ask for, so an
|
||||||
|
/// upgrade that adds one — SpelunkKey, ObjectStore — would never receive the items of that type made
|
||||||
|
/// before it. Starting over once is the only way to see them; applying is idempotent, which is what makes
|
||||||
|
/// the restart for a refused cursor safe as well. A state from before the list was kept has none and
|
||||||
|
/// starts over once for the same reason: it may already have missed some.
|
||||||
|
/// </remarks>
|
||||||
|
private async Task<StoredSyncState> ReadStateAsync(
|
||||||
|
Guid vaultId,
|
||||||
|
SyncReportBuilder report,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
var state = await syncState.ReadAsync(vaultId, cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
|
if (state.Cursor is not null && !string.Equals(state.PulledTypes, PulledTypes, StringComparison.Ordinal))
|
||||||
|
{
|
||||||
|
return await StartOverAsync(state, report, cancellationToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
// A pull from nothing that is not a restart — the vault's first, or a cache reset — sweeps as well.
|
||||||
|
// There is usually nothing to remove; the one time there is, it is the same case.
|
||||||
|
return state is { Cursor: null, SweepMark: null }
|
||||||
|
? await SweepingAsync(state, cancellationToken).ConfigureAwait(false)
|
||||||
|
: state;
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Forgets this vault's position and reads the log again from the beginning.
|
/// Forgets this vault's position and reads the log again from the beginning.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
@@ -234,44 +267,22 @@ public sealed class SyncEngine
|
|||||||
/// cache itself is the thing in doubt.
|
/// cache itself is the thing in doubt.
|
||||||
/// </para>
|
/// </para>
|
||||||
/// <para>
|
/// <para>
|
||||||
/// That leaves one gap, and it is worth naming rather than leaving to be discovered: once the server
|
/// A replay does not carry a deletion the server has collected — <c>DodoOptions.TombstoneRetentionDays</c>
|
||||||
/// starts collecting tombstones — <c>DodoOptions.TombstoneRetentionDays</c>, not implemented yet — a
|
/// — and that is the commonest refusal of all: a machine away longer than the retention. The rows it no
|
||||||
/// replay no longer carries a deletion older than the retention window. A machine that missed such a
|
/// longer hears about are the ones the replay does not mark, and are swept when it ends; see
|
||||||
/// delete and then had its cursor refused would keep the row. Nothing here can tell that from a row
|
/// <see cref="CarrySweepAsync"/>.
|
||||||
/// the server still has, so the answer when it matters will be to clear the mirror as well, not to
|
|
||||||
/// guess.
|
|
||||||
/// </para>
|
/// </para>
|
||||||
/// <para>
|
/// <para>
|
||||||
/// Written down before the replay begins, so a process that dies mid-replay starts the next one from
|
/// Written down before the replay begins, so a process that dies mid-replay starts the next one from
|
||||||
/// the beginning as well, rather than meeting the same refusal again.
|
/// the beginning as well, rather than meeting the same refusal again.
|
||||||
/// </para>
|
/// </para>
|
||||||
/// </remarks>
|
/// </remarks>
|
||||||
/// <summary>Where the pull should carry on from, which is the start when the types it reads have changed.</summary>
|
|
||||||
/// <remarks>
|
|
||||||
/// A cursor read under a different list of types is past changes of the types it did not ask for, so an
|
|
||||||
/// upgrade that adds one — SpelunkKey, ObjectStore — would never receive the items of that type made
|
|
||||||
/// before it. Starting over once is the only way to see them; applying is idempotent, which is what makes
|
|
||||||
/// the restart for a refused cursor safe as well. A state from before the list was kept has none and
|
|
||||||
/// starts over once for the same reason: it may already have missed some.
|
|
||||||
/// </remarks>
|
|
||||||
private async Task<StoredSyncState> ReadStateAsync(
|
|
||||||
Guid vaultId,
|
|
||||||
SyncReportBuilder report,
|
|
||||||
CancellationToken cancellationToken)
|
|
||||||
{
|
|
||||||
var state = await syncState.ReadAsync(vaultId, cancellationToken).ConfigureAwait(false);
|
|
||||||
|
|
||||||
return state.Cursor is not null && !string.Equals(state.PulledTypes, PulledTypes, StringComparison.Ordinal)
|
|
||||||
? await StartOverAsync(state, report, cancellationToken).ConfigureAwait(false)
|
|
||||||
: state;
|
|
||||||
}
|
|
||||||
|
|
||||||
private async Task<StoredSyncState> StartOverAsync(
|
private async Task<StoredSyncState> StartOverAsync(
|
||||||
StoredSyncState state,
|
StoredSyncState state,
|
||||||
SyncReportBuilder report,
|
SyncReportBuilder report,
|
||||||
CancellationToken cancellationToken)
|
CancellationToken cancellationToken)
|
||||||
{
|
{
|
||||||
var restarted = state with { Cursor = null };
|
var restarted = await SweepingAsync(state with { Cursor = null }, cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
await syncState.SaveAsync(restarted, cancellationToken).ConfigureAwait(false);
|
await syncState.SaveAsync(restarted, cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
@@ -280,6 +291,68 @@ public sealed class SyncEngine
|
|||||||
return restarted;
|
return restarted;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>The state with a sweep begun: a new mark, and the bound set from what the cache holds now.</summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// The mark is random rather than the time, because a sweep must never share one with the last: rows that
|
||||||
|
/// one marked would count as seen by this one. A clock can stand still, or go back.
|
||||||
|
/// </remarks>
|
||||||
|
private async Task<StoredSyncState> SweepingAsync(StoredSyncState state, CancellationToken cancellationToken) =>
|
||||||
|
state with
|
||||||
|
{
|
||||||
|
SweepMark = BitConverter.ToInt64(RandomNumberGenerator.GetBytes(sizeof(long))),
|
||||||
|
SweepSeenUpTo = await items.HighestSequenceAsync(state.VaultId, cancellationToken).ConfigureAwait(false),
|
||||||
|
};
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Marks what a pull from nothing has read, and at its end removes what it never did.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// The server forgets a deletion once it is past its retention. A machine that was away longer is
|
||||||
|
/// refused its cursor and reads everything again — and everything again is the items that still exist,
|
||||||
|
/// with nothing said about the ones that went. Without this their rows stayed here for good, and the
|
||||||
|
/// next edit of one put it back on the server for everybody.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// A pull from nothing can take more than one pass, so the mark is in the stored state, not in memory.
|
||||||
|
/// Every pull from nothing sweeps — the first one ever as well, where there is nothing to remove. What
|
||||||
|
/// is kept, and why, is <see cref="ItemStore.SweepAsync"/>'s to say.
|
||||||
|
/// </para>
|
||||||
|
/// </remarks>
|
||||||
|
private async Task<StoredSyncState> CarrySweepAsync(
|
||||||
|
Guid vaultId,
|
||||||
|
StoredSyncState state,
|
||||||
|
SyncPullResponse response,
|
||||||
|
SyncReportBuilder report,
|
||||||
|
CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
if (state.SweepMark is not { } mark)
|
||||||
|
{
|
||||||
|
return state;
|
||||||
|
}
|
||||||
|
|
||||||
|
var read = response.Changes.Where(change => reconcilers.ContainsKey(change.EntityType)).ToList();
|
||||||
|
|
||||||
|
await items.MarkSeenAsync(
|
||||||
|
vaultId,
|
||||||
|
mark,
|
||||||
|
read.Select(change => (change.EntityType, change.EntityId)).ToList(),
|
||||||
|
cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
|
var seenUpTo = read.Aggregate(state.SweepSeenUpTo, (highest, change) => Math.Max(highest, change.ChangeSequence));
|
||||||
|
|
||||||
|
if (response.HasMore)
|
||||||
|
{
|
||||||
|
return state with { SweepSeenUpTo = seenUpTo };
|
||||||
|
}
|
||||||
|
|
||||||
|
report.Pruned += await items
|
||||||
|
.SweepAsync(vaultId, ItemKinds.SyncedTypes, mark, seenUpTo, cancellationToken)
|
||||||
|
.ConfigureAwait(false);
|
||||||
|
|
||||||
|
return state with { SweepMark = null, SweepSeenUpTo = 0 };
|
||||||
|
}
|
||||||
|
|
||||||
private StoredSyncState Record(
|
private StoredSyncState Record(
|
||||||
Guid vaultId,
|
Guid vaultId,
|
||||||
StoredSyncState state,
|
StoredSyncState state,
|
||||||
@@ -794,6 +867,26 @@ public sealed class SyncEngine
|
|||||||
|
|
||||||
if (failure is null)
|
if (failure is null)
|
||||||
{
|
{
|
||||||
|
// An edit of a version this machine once saw means the server had the item and has since let go
|
||||||
|
// of it: deleted elsewhere, and the deletion collected past its retention before this machine
|
||||||
|
// heard of it. The edit is the user's work and goes back, but not silently — somebody deleted
|
||||||
|
// this on purpose. A create has no expected version and is the ordinary case.
|
||||||
|
if (operation.ExpectedVersion is not null)
|
||||||
|
{
|
||||||
|
await conflicts.RecordAsync(
|
||||||
|
vaultId,
|
||||||
|
operation.EntityType,
|
||||||
|
operation.EntityId,
|
||||||
|
ConflictKind.RecreatedAfterCollection,
|
||||||
|
ConflictDetails.Encode(
|
||||||
|
"This was deleted elsewhere while this machine had unsaved changes to it, long enough ago "
|
||||||
|
+ "that the server no longer remembered the deletion. The changes were kept and the item "
|
||||||
|
+ "was put back."),
|
||||||
|
cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
|
report.Recreated++;
|
||||||
|
}
|
||||||
|
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -84,6 +84,17 @@ public sealed record SyncOptions
|
|||||||
/// changed anything, which is otherwise the sort of thing that looks like a fault.
|
/// changed anything, which is otherwise the sort of thing that looks like a fault.
|
||||||
/// </para>
|
/// </para>
|
||||||
/// </param>
|
/// </param>
|
||||||
|
/// <param name="PushedLogEntries">How many of <paramref name="Pushed"/> were log entries.</param>
|
||||||
|
/// <param name="PulledLogEntries">How many of <paramref name="Pulled"/> were log entries.</param>
|
||||||
|
/// <param name="Recreated">
|
||||||
|
/// Items this machine had edited that somebody else had deleted so long ago the server no longer
|
||||||
|
/// remembered it. The edit was kept and the item sent again under its own name — and the user is told,
|
||||||
|
/// because the deletion was somebody's decision.
|
||||||
|
/// </param>
|
||||||
|
/// <param name="Pruned">
|
||||||
|
/// Items removed from this machine because a read from the beginning found the server no longer has them:
|
||||||
|
/// deleted elsewhere while this machine was away for longer than the server keeps deletions.
|
||||||
|
/// </param>
|
||||||
public sealed record SyncReport(
|
public sealed record SyncReport(
|
||||||
Guid VaultId,
|
Guid VaultId,
|
||||||
int Pulled,
|
int Pulled,
|
||||||
@@ -99,11 +110,13 @@ public sealed record SyncReport(
|
|||||||
bool RoundsExhausted,
|
bool RoundsExhausted,
|
||||||
bool ResyncedFromStart,
|
bool ResyncedFromStart,
|
||||||
int PushedLogEntries = 0,
|
int PushedLogEntries = 0,
|
||||||
int PulledLogEntries = 0)
|
int PulledLogEntries = 0,
|
||||||
|
int Recreated = 0,
|
||||||
|
int Pruned = 0)
|
||||||
{
|
{
|
||||||
/// <summary>Whether anything happened that a user should be told about.</summary>
|
/// <summary>Whether anything happened that a user should be told about.</summary>
|
||||||
public bool NeedsAttention =>
|
public bool NeedsAttention =>
|
||||||
Resurrected > 0 || DeletesAbandoned > 0 || Parked > 0 || Unreadable > 0 || RekeyRequired;
|
Resurrected > 0 || Recreated > 0 || DeletesAbandoned > 0 || Parked > 0 || Unreadable > 0 || RekeyRequired;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Operations the server accepted that were somebody's own work rather than a log entry.
|
/// Operations the server accepted that were somebody's own work rather than a log entry.
|
||||||
@@ -160,6 +173,10 @@ internal sealed class SyncReportBuilder(Guid vaultId)
|
|||||||
|
|
||||||
internal bool ResyncedFromStart { get; set; }
|
internal bool ResyncedFromStart { get; set; }
|
||||||
|
|
||||||
|
internal int Recreated { get; set; }
|
||||||
|
|
||||||
|
internal int Pruned { get; set; }
|
||||||
|
|
||||||
internal SyncReport Build() =>
|
internal SyncReport Build() =>
|
||||||
new(
|
new(
|
||||||
vaultId,
|
vaultId,
|
||||||
@@ -176,7 +193,9 @@ internal sealed class SyncReportBuilder(Guid vaultId)
|
|||||||
RoundsExhausted,
|
RoundsExhausted,
|
||||||
ResyncedFromStart,
|
ResyncedFromStart,
|
||||||
PushedLogEntries,
|
PushedLogEntries,
|
||||||
PulledLogEntries);
|
PulledLogEntries,
|
||||||
|
Recreated,
|
||||||
|
Pruned);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>The record written to the conflict log when a merge had to override something.</summary>
|
/// <summary>The record written to the conflict log when a merge had to override something.</summary>
|
||||||
|
|||||||
@@ -19,10 +19,21 @@ namespace DodoSSH.Domain.Sync;
|
|||||||
/// The cursor also carries its vault id, so a legitimately-issued cursor for one vault cannot be
|
/// The cursor also carries its vault id, so a legitimately-issued cursor for one vault cannot be
|
||||||
/// replayed against another. See ADR 0003.
|
/// replayed against another. See ADR 0003.
|
||||||
/// </para>
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// And the vault's sync horizon when it was issued — the last deletion the server had forgotten. A cursor
|
||||||
|
/// behind the horizon is refused when it is older than the collection, because its holder may still have the
|
||||||
|
/// deleted item; it is fine when it is newer, because then it belongs to a read from the beginning that
|
||||||
|
/// started after the collection, whose early pages are behind the horizon by nature. Position alone cannot
|
||||||
|
/// tell those two apart. A cursor issued under no horizon keeps the first format, so a server from before
|
||||||
|
/// this still reads it.
|
||||||
|
/// </para>
|
||||||
/// </remarks>
|
/// </remarks>
|
||||||
public static class SyncCursor
|
public static class SyncCursor
|
||||||
{
|
{
|
||||||
private const string Version = "v1";
|
private const string Version = "v1";
|
||||||
|
|
||||||
|
/// <summary>The format that carries a horizon as well.</summary>
|
||||||
|
private const string HorizonVersion = "v2";
|
||||||
private const char Separator = '|';
|
private const char Separator = '|';
|
||||||
|
|
||||||
/// <summary>Length of the truncated HMAC tag. 128 bits is ample for a non-secret position.</summary>
|
/// <summary>Length of the truncated HMAC tag. 128 bits is ample for a non-secret position.</summary>
|
||||||
@@ -37,7 +48,17 @@ public static class SyncCursor
|
|||||||
/// <param name="signingKey">Deployment cursor signing key, at least 32 bytes.</param>
|
/// <param name="signingKey">Deployment cursor signing key, at least 32 bytes.</param>
|
||||||
/// <param name="vaultId">Vault the cursor belongs to.</param>
|
/// <param name="vaultId">Vault the cursor belongs to.</param>
|
||||||
/// <param name="sequence">Last change sequence the client has consumed.</param>
|
/// <param name="sequence">Last change sequence the client has consumed.</param>
|
||||||
public static string Encode(ReadOnlySpan<byte> signingKey, Guid vaultId, long sequence)
|
public static string Encode(ReadOnlySpan<byte> signingKey, Guid vaultId, long sequence) =>
|
||||||
|
Encode(signingKey, vaultId, sequence, horizon: 0);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Encodes a position, with the sync horizon it was read under.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="signingKey">Deployment cursor signing key, at least 32 bytes.</param>
|
||||||
|
/// <param name="vaultId">Vault the cursor belongs to.</param>
|
||||||
|
/// <param name="sequence">Last change sequence the client has consumed.</param>
|
||||||
|
/// <param name="horizon">The vault's sync horizon before the read that produced this position.</param>
|
||||||
|
public static string Encode(ReadOnlySpan<byte> signingKey, Guid vaultId, long sequence, long horizon)
|
||||||
{
|
{
|
||||||
RequireKey(signingKey);
|
RequireKey(signingKey);
|
||||||
|
|
||||||
@@ -46,7 +67,12 @@ public static class SyncCursor
|
|||||||
throw new ArgumentOutOfRangeException(nameof(sequence), sequence, "Sequence must not be negative.");
|
throw new ArgumentOutOfRangeException(nameof(sequence), sequence, "Sequence must not be negative.");
|
||||||
}
|
}
|
||||||
|
|
||||||
var payload = Encoding.UTF8.GetBytes(FormatPayload(vaultId, sequence));
|
if (horizon < 0)
|
||||||
|
{
|
||||||
|
throw new ArgumentOutOfRangeException(nameof(horizon), horizon, "Horizon must not be negative.");
|
||||||
|
}
|
||||||
|
|
||||||
|
var payload = Encoding.UTF8.GetBytes(FormatPayload(vaultId, sequence, horizon));
|
||||||
|
|
||||||
var buffer = new byte[payload.Length + TagLength];
|
var buffer = new byte[payload.Length + TagLength];
|
||||||
payload.CopyTo(buffer, 0);
|
payload.CopyTo(buffer, 0);
|
||||||
@@ -70,10 +96,27 @@ public static class SyncCursor
|
|||||||
ReadOnlySpan<byte> signingKey,
|
ReadOnlySpan<byte> signingKey,
|
||||||
string? cursor,
|
string? cursor,
|
||||||
Guid expectedVaultId,
|
Guid expectedVaultId,
|
||||||
out long sequence)
|
out long sequence) =>
|
||||||
|
TryDecode(signingKey, cursor, expectedVaultId, out sequence, out _);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Decodes and verifies a cursor, with the horizon it was issued under — zero for one issued under none.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="signingKey">Deployment cursor signing key.</param>
|
||||||
|
/// <param name="cursor">The cursor to verify.</param>
|
||||||
|
/// <param name="expectedVaultId">Vault the request is scoped to.</param>
|
||||||
|
/// <param name="sequence">The decoded sequence, when verification succeeds.</param>
|
||||||
|
/// <param name="horizon">The decoded horizon, when verification succeeds.</param>
|
||||||
|
public static bool TryDecode(
|
||||||
|
ReadOnlySpan<byte> signingKey,
|
||||||
|
string? cursor,
|
||||||
|
Guid expectedVaultId,
|
||||||
|
out long sequence,
|
||||||
|
out long horizon)
|
||||||
{
|
{
|
||||||
RequireKey(signingKey);
|
RequireKey(signingKey);
|
||||||
sequence = 0;
|
sequence = 0;
|
||||||
|
horizon = 0;
|
||||||
|
|
||||||
if (string.IsNullOrEmpty(cursor))
|
if (string.IsNullOrEmpty(cursor))
|
||||||
{
|
{
|
||||||
@@ -97,17 +140,26 @@ public static class SyncCursor
|
|||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
return TryParsePayload(payload, expectedVaultId, out sequence);
|
return TryParsePayload(payload, expectedVaultId, out sequence, out horizon);
|
||||||
}
|
}
|
||||||
|
|
||||||
private static string FormatPayload(Guid vaultId, long sequence) =>
|
private static string FormatPayload(Guid vaultId, long sequence, long horizon) =>
|
||||||
string.Create(
|
horizon == 0
|
||||||
CultureInfo.InvariantCulture,
|
? string.Create(
|
||||||
$"{Version}{Separator}{vaultId:D}{Separator}{sequence}");
|
CultureInfo.InvariantCulture,
|
||||||
|
$"{Version}{Separator}{vaultId:D}{Separator}{sequence}")
|
||||||
|
: string.Create(
|
||||||
|
CultureInfo.InvariantCulture,
|
||||||
|
$"{HorizonVersion}{Separator}{vaultId:D}{Separator}{sequence}{Separator}{horizon}");
|
||||||
|
|
||||||
private static bool TryParsePayload(ReadOnlySpan<byte> payload, Guid expectedVaultId, out long sequence)
|
private static bool TryParsePayload(
|
||||||
|
ReadOnlySpan<byte> payload,
|
||||||
|
Guid expectedVaultId,
|
||||||
|
out long sequence,
|
||||||
|
out long horizon)
|
||||||
{
|
{
|
||||||
sequence = 0;
|
sequence = 0;
|
||||||
|
horizon = 0;
|
||||||
|
|
||||||
string text;
|
string text;
|
||||||
try
|
try
|
||||||
@@ -122,12 +174,14 @@ public static class SyncCursor
|
|||||||
// Tagged, so the shape is ours — but parse defensively anyway, because a key rotation
|
// Tagged, so the shape is ours — but parse defensively anyway, because a key rotation
|
||||||
// could make an old format verify against a new expectation.
|
// could make an old format verify against a new expectation.
|
||||||
var parts = text.Split(Separator);
|
var parts = text.Split(Separator);
|
||||||
if (parts.Length != 3)
|
var expectedParts = parts[0] switch
|
||||||
{
|
{
|
||||||
return false;
|
Version => 3,
|
||||||
}
|
HorizonVersion => 4,
|
||||||
|
_ => 0,
|
||||||
|
};
|
||||||
|
|
||||||
if (!string.Equals(parts[0], Version, StringComparison.Ordinal))
|
if (parts.Length != expectedParts)
|
||||||
{
|
{
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
@@ -137,8 +191,13 @@ public static class SyncCursor
|
|||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
return long.TryParse(parts[2], NumberStyles.None, CultureInfo.InvariantCulture, out sequence)
|
if (!long.TryParse(parts[2], NumberStyles.None, CultureInfo.InvariantCulture, out sequence))
|
||||||
&& sequence >= 0;
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
return expectedParts == 3
|
||||||
|
|| long.TryParse(parts[3], NumberStyles.None, CultureInfo.InvariantCulture, out horizon);
|
||||||
}
|
}
|
||||||
|
|
||||||
private static void ComputeTag(
|
private static void ComputeTag(
|
||||||
|
|||||||
@@ -52,6 +52,16 @@ public sealed class Vault
|
|||||||
/// <summary>Soft-delete marker.</summary>
|
/// <summary>Soft-delete marker.</summary>
|
||||||
public DateTimeOffset? DeletedAtUtc { get; set; }
|
public DateTimeOffset? DeletedAtUtc { get; set; }
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The highest change-log sequence whose tombstone has been collected from this vault.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// A cursor below it may have been issued before a deletion this server no longer records, so a pull
|
||||||
|
/// from it is refused and the client starts over — see <c>TombstoneCollector</c> and ADR 0003. Zero until
|
||||||
|
/// anything has been collected.
|
||||||
|
/// </remarks>
|
||||||
|
public long SyncHorizon { get; set; }
|
||||||
|
|
||||||
/// <summary>Wrapped vault keys, one per recipient per generation.</summary>
|
/// <summary>Wrapped vault keys, one per recipient per generation.</summary>
|
||||||
public ICollection<VaultKeyGrant> KeyGrants { get; } = [];
|
public ICollection<VaultKeyGrant> KeyGrants { get; } = [];
|
||||||
|
|
||||||
|
|||||||
+2004
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,31 @@
|
|||||||
|
using Microsoft.EntityFrameworkCore.Migrations;
|
||||||
|
|
||||||
|
#nullable disable
|
||||||
|
|
||||||
|
namespace DodoSSH.Infrastructure.Migrations
|
||||||
|
{
|
||||||
|
/// <inheritdoc />
|
||||||
|
public partial class AddVaultSyncHorizon : Migration
|
||||||
|
{
|
||||||
|
/// <inheritdoc />
|
||||||
|
protected override void Up(MigrationBuilder migrationBuilder)
|
||||||
|
{
|
||||||
|
migrationBuilder.AddColumn<long>(
|
||||||
|
name: "sync_horizon",
|
||||||
|
schema: "dodo",
|
||||||
|
table: "vault",
|
||||||
|
type: "bigint",
|
||||||
|
nullable: false,
|
||||||
|
defaultValue: 0L);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
protected override void Down(MigrationBuilder migrationBuilder)
|
||||||
|
{
|
||||||
|
migrationBuilder.DropColumn(
|
||||||
|
name: "sync_horizon",
|
||||||
|
schema: "dodo",
|
||||||
|
table: "vault");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -750,6 +750,10 @@ namespace DodoSSH.Infrastructure.Migrations
|
|||||||
.HasColumnType("boolean")
|
.HasColumnType("boolean")
|
||||||
.HasColumnName("rekey_required");
|
.HasColumnName("rekey_required");
|
||||||
|
|
||||||
|
b.Property<long>("SyncHorizon")
|
||||||
|
.HasColumnType("bigint")
|
||||||
|
.HasColumnName("sync_horizon");
|
||||||
|
|
||||||
b.Property<Guid?>("TeamId")
|
b.Property<Guid?>("TeamId")
|
||||||
.HasColumnType("uuid")
|
.HasColumnType("uuid")
|
||||||
.HasColumnName("team_id");
|
.HasColumnName("team_id");
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
using System.Net;
|
using System.Net;
|
||||||
using System.Net.Http.Json;
|
using System.Net.Http.Json;
|
||||||
|
using DodoSSH.Api.Features.Sync;
|
||||||
using DodoSSH.Contracts;
|
using DodoSSH.Contracts;
|
||||||
using DodoSSH.Domain;
|
using DodoSSH.Domain;
|
||||||
using DodoSSH.Infrastructure;
|
using DodoSSH.Infrastructure;
|
||||||
@@ -612,6 +613,209 @@ public sealed class SyncEndpointTests(ApiFixture fixture)
|
|||||||
tombstone.PlaintextFields.ShouldBeNull();
|
tombstone.PlaintextFields.ShouldBeNull();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ---- Tombstone collection (ADR 0003) ----
|
||||||
|
|
||||||
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// A tombstone past its retention goes, with its change-log entries — and a client whose cursor is from
|
||||||
|
/// before it is refused, because it would otherwise never hear of the deletion. A client already past it
|
||||||
|
/// carries on, and a pull from nothing simply no longer offers the item.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// The retention is crossed by moving the deletion back in time rather than by waiting it out.
|
||||||
|
/// </para>
|
||||||
|
/// </remarks>
|
||||||
|
[Fact]
|
||||||
|
public async Task AnExpiredTombstone_IsCollected_AndACursorFromBeforeIt_IsRefused()
|
||||||
|
{
|
||||||
|
var (subject, vaultId) = await SeedUserWithVaultAsync();
|
||||||
|
var client = fixture.CreateClientFor(subject);
|
||||||
|
|
||||||
|
var create = NewCreateBatch();
|
||||||
|
var entityId = create.Operations[0].EntityId;
|
||||||
|
await client.PostContractAsync(PushUrl(vaultId), create);
|
||||||
|
|
||||||
|
var before = await PullAsync(client, vaultId, null);
|
||||||
|
|
||||||
|
await client.PostContractAsync(PushUrl(vaultId), new SyncPushRequest(
|
||||||
|
[
|
||||||
|
new SyncPushOperation(
|
||||||
|
Guid.CreateVersion7(), SyncEntityType.Host, entityId, SyncOperation.Delete,
|
||||||
|
ExpectedVersion: 1, Payload: null, PlaintextFields: null),
|
||||||
|
]));
|
||||||
|
|
||||||
|
var after = await PullAsync(client, vaultId, before.NextCursor);
|
||||||
|
|
||||||
|
await AgeTheDeletionAsync(entityId);
|
||||||
|
await CollectAsync();
|
||||||
|
|
||||||
|
var refused = await client.PostContractAsync(PullUrl(vaultId), new SyncPullRequest(before.NextCursor, null, null));
|
||||||
|
refused.StatusCode.ShouldBe(HttpStatusCode.BadRequest);
|
||||||
|
(await refused.Content.ReadProblemAsync()).ShouldNotBeNull().Code.ShouldBe(ProblemCodes.InvalidCursor);
|
||||||
|
|
||||||
|
(await PullAsync(client, vaultId, after.NextCursor)).Changes.ShouldBeEmpty("a cursor past the deletion is fine");
|
||||||
|
|
||||||
|
(await PullAsync(client, vaultId, null)).Changes.ShouldNotContain(change => change.EntityId == entityId);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// The client's way back after a refusal is to read from the beginning — and that read's own cursors are
|
||||||
|
/// behind the horizon whenever an item older than the collected deletion is still there, which is nearly
|
||||||
|
/// always. Refused on position alone, the second page sent the read back to the start, every time, for
|
||||||
|
/// good. The cursor carries the horizon it was issued under, and one issued since the collection passes.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// One item a page, so the first page's cursor is the old item's position, well behind the horizon.
|
||||||
|
/// </para>
|
||||||
|
/// </remarks>
|
||||||
|
[Fact]
|
||||||
|
public async Task AReadFromTheBeginning_AfterACollection_IsNotRefusedOnItsOwnPages()
|
||||||
|
{
|
||||||
|
var (subject, vaultId) = await SeedUserWithVaultAsync();
|
||||||
|
var client = fixture.CreateClientFor(subject);
|
||||||
|
|
||||||
|
var old = NewCreateBatch();
|
||||||
|
await client.PostContractAsync(PushUrl(vaultId), old);
|
||||||
|
await client.PostContractAsync(PushUrl(vaultId), NewCreateBatch());
|
||||||
|
|
||||||
|
var doomed = NewCreateBatch();
|
||||||
|
var entityId = doomed.Operations[0].EntityId;
|
||||||
|
await client.PostContractAsync(PushUrl(vaultId), doomed);
|
||||||
|
|
||||||
|
await client.PostContractAsync(PushUrl(vaultId), new SyncPushRequest(
|
||||||
|
[
|
||||||
|
new SyncPushOperation(
|
||||||
|
Guid.CreateVersion7(), SyncEntityType.Host, entityId, SyncOperation.Delete,
|
||||||
|
ExpectedVersion: 1, Payload: null, PlaintextFields: null),
|
||||||
|
]));
|
||||||
|
|
||||||
|
await AgeTheDeletionAsync(entityId);
|
||||||
|
await CollectAsync();
|
||||||
|
|
||||||
|
var seen = new List<Guid>();
|
||||||
|
string? cursor = null;
|
||||||
|
|
||||||
|
for (var page = 0; page < 5; page++)
|
||||||
|
{
|
||||||
|
var response = await client.PostContractAsync(PullUrl(vaultId), new SyncPullRequest(cursor, 1, null));
|
||||||
|
|
||||||
|
response.StatusCode.ShouldBe(HttpStatusCode.OK, $"page {page} was refused");
|
||||||
|
|
||||||
|
var body = (await response.Content.ReadContractAsync<SyncPullResponse>()).ShouldNotBeNull();
|
||||||
|
seen.AddRange(body.Changes.Select(change => change.EntityId));
|
||||||
|
cursor = body.NextCursor;
|
||||||
|
|
||||||
|
if (!body.HasMore)
|
||||||
|
{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
seen.Count.ShouldBe(2);
|
||||||
|
seen.ShouldContain(old.Operations[0].EntityId);
|
||||||
|
seen.ShouldNotContain(entityId);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <remarks>
|
||||||
|
/// The other half of the retention: a deletion younger than it stays, entries and all, so a machine that
|
||||||
|
/// has been away a week still hears of it.
|
||||||
|
/// </remarks>
|
||||||
|
[Fact]
|
||||||
|
public async Task ATombstoneInsideItsRetention_IsKept()
|
||||||
|
{
|
||||||
|
var (subject, vaultId) = await SeedUserWithVaultAsync();
|
||||||
|
var client = fixture.CreateClientFor(subject);
|
||||||
|
|
||||||
|
var create = NewCreateBatch();
|
||||||
|
var entityId = create.Operations[0].EntityId;
|
||||||
|
await client.PostContractAsync(PushUrl(vaultId), create);
|
||||||
|
|
||||||
|
var before = await PullAsync(client, vaultId, null);
|
||||||
|
|
||||||
|
await client.PostContractAsync(PushUrl(vaultId), new SyncPushRequest(
|
||||||
|
[
|
||||||
|
new SyncPushOperation(
|
||||||
|
Guid.CreateVersion7(), SyncEntityType.Host, entityId, SyncOperation.Delete,
|
||||||
|
ExpectedVersion: 1, Payload: null, PlaintextFields: null),
|
||||||
|
]));
|
||||||
|
|
||||||
|
await CollectAsync();
|
||||||
|
|
||||||
|
var later = await PullAsync(client, vaultId, before.NextCursor);
|
||||||
|
|
||||||
|
later.Changes.ShouldHaveSingleItem().Operation.ShouldBe(SyncOperation.Delete);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <remarks>
|
||||||
|
/// An item edited many times kept an entry per edit for ever, and a client starting from nothing walked
|
||||||
|
/// every one of them. After a collection only the newest is left — and a client that held a cursor from
|
||||||
|
/// between the edits still gets the item, because what it had not seen yet is exactly that newest entry.
|
||||||
|
/// </remarks>
|
||||||
|
[Fact]
|
||||||
|
public async Task SupersededEntries_AreCompactedAway_WithoutCostingAnyClientTheItem()
|
||||||
|
{
|
||||||
|
var (subject, vaultId) = await SeedUserWithVaultAsync();
|
||||||
|
var client = fixture.CreateClientFor(subject);
|
||||||
|
|
||||||
|
var create = NewCreateBatch();
|
||||||
|
var entityId = create.Operations[0].EntityId;
|
||||||
|
await client.PostContractAsync(PushUrl(vaultId), create);
|
||||||
|
|
||||||
|
var between = await PullAsync(client, vaultId, null);
|
||||||
|
|
||||||
|
await client.PostContractAsync(PushUrl(vaultId), new SyncPushRequest(
|
||||||
|
[
|
||||||
|
NewOperation(entityId, expectedVersion: 1, envelope: [2, 2, 2]),
|
||||||
|
]));
|
||||||
|
|
||||||
|
await client.PostContractAsync(PushUrl(vaultId), new SyncPushRequest(
|
||||||
|
[
|
||||||
|
NewOperation(entityId, expectedVersion: 2, envelope: [3, 3, 3]),
|
||||||
|
]));
|
||||||
|
|
||||||
|
await CollectAsync();
|
||||||
|
|
||||||
|
await using (var scope = fixture.CreateScope())
|
||||||
|
{
|
||||||
|
var database = scope.ServiceProvider.GetRequiredService<DodoDbContext>();
|
||||||
|
|
||||||
|
(await database.VaultChanges.CountAsync(c => c.EntityId == entityId, TestContext.Current.CancellationToken))
|
||||||
|
.ShouldBe(1);
|
||||||
|
}
|
||||||
|
|
||||||
|
(await PullAsync(client, vaultId, between.NextCursor)).Changes.ShouldHaveSingleItem().Version.ShouldBe(3);
|
||||||
|
(await PullAsync(client, vaultId, null)).Changes.Single(c => c.EntityId == entityId).Version.ShouldBe(3);
|
||||||
|
}
|
||||||
|
|
||||||
|
private static async Task<SyncPullResponse> PullAsync(HttpClient client, Guid vaultId, string? cursor)
|
||||||
|
{
|
||||||
|
var response = await client.PostContractAsync(PullUrl(vaultId), new SyncPullRequest(cursor, null, null));
|
||||||
|
|
||||||
|
response.StatusCode.ShouldBe(HttpStatusCode.OK);
|
||||||
|
|
||||||
|
return (await response.Content.ReadContractAsync<SyncPullResponse>()).ShouldNotBeNull();
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Moves a deletion back past the retention, which is quicker than waiting ninety days.</summary>
|
||||||
|
private async Task AgeTheDeletionAsync(Guid entityId)
|
||||||
|
{
|
||||||
|
await using var scope = fixture.CreateScope();
|
||||||
|
var database = scope.ServiceProvider.GetRequiredService<DodoDbContext>();
|
||||||
|
|
||||||
|
var host = await database.Hosts.SingleAsync(h => h.Id == entityId, TestContext.Current.CancellationToken);
|
||||||
|
host.DeletedAtUtc = host.DeletedAtUtc!.Value.AddDays(-91);
|
||||||
|
|
||||||
|
await database.SaveChangesAsync(TestContext.Current.CancellationToken);
|
||||||
|
}
|
||||||
|
|
||||||
|
private async Task CollectAsync()
|
||||||
|
{
|
||||||
|
await using var scope = fixture.CreateScope();
|
||||||
|
|
||||||
|
await scope.ServiceProvider.GetRequiredService<TombstoneCollector>().RunOnceAsync(TestContext.Current.CancellationToken);
|
||||||
|
}
|
||||||
|
|
||||||
// ---- Batch limits ----
|
// ---- Batch limits ----
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
|
|||||||
@@ -102,6 +102,7 @@ internal sealed class FakeVaultServer : ISyncApi
|
|||||||
SyncPullRequest request,
|
SyncPullRequest request,
|
||||||
CancellationToken cancellationToken)
|
CancellationToken cancellationToken)
|
||||||
{
|
{
|
||||||
|
var issuedUnder = horizon;
|
||||||
var after = DecodeCursor(request.Cursor);
|
var after = DecodeCursor(request.Cursor);
|
||||||
var limit = Math.Clamp(request.Limit ?? MaxPullLimit, 1, MaxPullLimit);
|
var limit = Math.Clamp(request.Limit ?? MaxPullLimit, 1, MaxPullLimit);
|
||||||
|
|
||||||
@@ -128,7 +129,7 @@ internal sealed class FakeVaultServer : ISyncApi
|
|||||||
|
|
||||||
return Task.FromResult(new SyncPullResponse(
|
return Task.FromResult(new SyncPullResponse(
|
||||||
[.. page.Select(entry => Hydrate(entry))],
|
[.. page.Select(entry => Hydrate(entry))],
|
||||||
EncodeCursor(next),
|
EncodeCursor(next, issuedUnder),
|
||||||
hasMore,
|
hasMore,
|
||||||
Now,
|
Now,
|
||||||
KeyGeneration));
|
KeyGeneration));
|
||||||
@@ -206,7 +207,25 @@ internal sealed class FakeVaultServer : ISyncApi
|
|||||||
internal Row? Find(Guid entityId, SyncEntityType entityType = SyncEntityType.Host) =>
|
internal Row? Find(Guid entityId, SyncEntityType entityType = SyncEntityType.Host) =>
|
||||||
rows.TryGetValue((entityType, entityId), out var row) ? row : null;
|
rows.TryGetValue((entityType, entityId), out var row) ? row : null;
|
||||||
|
|
||||||
private long Head => log.Count == 0 ? 0 : log[^1].Sequence;
|
/// <summary>
|
||||||
|
/// Forgets every deletion, as the real server does once one is past its retention: the tombstone and
|
||||||
|
/// its log entries go, and a cursor from before them is refused from then on.
|
||||||
|
/// </summary>
|
||||||
|
internal void CollectTombstones()
|
||||||
|
{
|
||||||
|
foreach (var (key, row) in rows.Where(entry => entry.Value.IsDeleted).ToList())
|
||||||
|
{
|
||||||
|
rows.Remove(key);
|
||||||
|
log.RemoveAll(entry => entry.EntityType == key.Type && entry.EntityId == key.EntityId);
|
||||||
|
horizon = Math.Max(horizon, row.ChangeSequence);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>The newest sequence ever issued, which a collection must not take back.</summary>
|
||||||
|
private long Head => Math.Max(log.Count == 0 ? 0 : log[^1].Sequence, horizon);
|
||||||
|
|
||||||
|
/// <summary>The last deletion forgotten; see <see cref="CollectTombstones"/>.</summary>
|
||||||
|
private long horizon;
|
||||||
|
|
||||||
// ---- The decision table ----
|
// ---- The decision table ----
|
||||||
|
|
||||||
@@ -481,8 +500,12 @@ internal sealed class FakeVaultServer : ISyncApi
|
|||||||
/// rejects. The real server HMAC-tags them; the property that matters to the client is only that it
|
/// rejects. The real server HMAC-tags them; the property that matters to the client is only that it
|
||||||
/// must round-trip what it is given.
|
/// must round-trip what it is given.
|
||||||
/// </remarks>
|
/// </remarks>
|
||||||
private static string EncodeCursor(long sequence) =>
|
/// <remarks>
|
||||||
"fake-v1:" + sequence.ToString(CultureInfo.InvariantCulture);
|
/// Carries the horizon it was read under, as the real one does, so a read from the beginning that started
|
||||||
|
/// after a collection is not refused on its own early pages. See <c>DodoSSH.Domain.Sync.SyncCursor</c>.
|
||||||
|
/// </remarks>
|
||||||
|
private static string EncodeCursor(long sequence, long issuedUnder = 0) =>
|
||||||
|
string.Create(CultureInfo.InvariantCulture, $"fake-v1:{sequence}@{issuedUnder}");
|
||||||
|
|
||||||
private long DecodeCursor(string? cursor)
|
private long DecodeCursor(string? cursor)
|
||||||
{
|
{
|
||||||
@@ -502,12 +525,26 @@ internal sealed class FakeVaultServer : ISyncApi
|
|||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var at = cursor.IndexOf('@', StringComparison.Ordinal);
|
||||||
|
|
||||||
if (!cursor.StartsWith("fake-v1:", StringComparison.Ordinal)
|
if (!cursor.StartsWith("fake-v1:", StringComparison.Ordinal)
|
||||||
|| !long.TryParse(cursor.AsSpan(8), CultureInfo.InvariantCulture, out var sequence))
|
|| at < 0
|
||||||
|
|| !long.TryParse(cursor.AsSpan(8, at - 8), CultureInfo.InvariantCulture, out var sequence)
|
||||||
|
|| !long.TryParse(cursor.AsSpan(at + 1), CultureInfo.InvariantCulture, out var issuedUnder))
|
||||||
{
|
{
|
||||||
throw new InvalidOperationException($"A client sent a cursor it should not have: '{cursor}'.");
|
throw new InvalidOperationException($"A client sent a cursor it should not have: '{cursor}'.");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (sequence < horizon && issuedUnder < horizon)
|
||||||
|
{
|
||||||
|
// From before a deletion this server no longer remembers, so it could never be told of it.
|
||||||
|
throw new DodoSshApiException(
|
||||||
|
HttpStatusCode.BadRequest,
|
||||||
|
ProblemCodes.InvalidCursor,
|
||||||
|
"The server returned 400: The sync cursor is not valid for this vault. "
|
||||||
|
+ "Resync from the beginning.");
|
||||||
|
}
|
||||||
|
|
||||||
return sequence;
|
return sequence;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,142 @@
|
|||||||
|
using DodoSSH.Client.Storage;
|
||||||
|
using static DodoSSH.Client.Sync.Tests.SyncHarness;
|
||||||
|
|
||||||
|
namespace DodoSSH.Client.Sync.Tests;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A machine away for longer than the server keeps deletions (ADR 0003).
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// The server forgets a deletion once it is past its retention and refuses any cursor from before it. The
|
||||||
|
/// machine then reads everything again — which says nothing about what is gone — so what it held and did
|
||||||
|
/// not read again is what has to go. Except the user's own unsent work, which is kept and said out loud.
|
||||||
|
/// </remarks>
|
||||||
|
public sealed class TombstoneCollectionTests
|
||||||
|
{
|
||||||
|
[Fact]
|
||||||
|
public async Task AnItemDeletedAndForgottenWhileAway_IsRemovedByTheReadFromTheBeginning()
|
||||||
|
{
|
||||||
|
using var harness = await CreateAsync();
|
||||||
|
|
||||||
|
var gone = await harness.Second.CreateAsync(Host("old-db"));
|
||||||
|
var kept = await harness.Second.CreateAsync(Host("prod-db"));
|
||||||
|
await harness.Second.SyncAsync();
|
||||||
|
await harness.First.SyncAsync();
|
||||||
|
|
||||||
|
await harness.Second.DeleteAsync(gone);
|
||||||
|
await harness.Second.SyncAsync();
|
||||||
|
harness.Server.CollectTombstones();
|
||||||
|
|
||||||
|
var report = await harness.First.SyncAsync();
|
||||||
|
|
||||||
|
report.ResyncedFromStart.ShouldBeTrue();
|
||||||
|
report.Pruned.ShouldBe(1);
|
||||||
|
(await harness.First.ListAsync()).Items.Select(item => item.EntityId).ShouldBe([kept]);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task TheOnlyItemInAVault_IsRemovedAsWell()
|
||||||
|
{
|
||||||
|
// The read from the beginning brings back nothing at all, so a sweep bounded by what it read would
|
||||||
|
// reach nothing either. It is bounded by what the cache held when it started.
|
||||||
|
using var harness = await CreateAsync();
|
||||||
|
|
||||||
|
var gone = await harness.Second.CreateAsync(Host("old-db"));
|
||||||
|
await harness.Second.SyncAsync();
|
||||||
|
await harness.First.SyncAsync();
|
||||||
|
|
||||||
|
await harness.Second.DeleteAsync(gone);
|
||||||
|
await harness.Second.SyncAsync();
|
||||||
|
harness.Server.CollectTombstones();
|
||||||
|
|
||||||
|
var report = await harness.First.SyncAsync();
|
||||||
|
|
||||||
|
report.Pruned.ShouldBe(1);
|
||||||
|
(await harness.First.ListAsync()).Items.ShouldBeEmpty();
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task AnItemWithAnUnsentEdit_IsKept_PutBack_AndSaidSo()
|
||||||
|
{
|
||||||
|
using var harness = await CreateAsync();
|
||||||
|
|
||||||
|
var entityId = await harness.Second.CreateAsync(Host("old-db"));
|
||||||
|
await harness.Second.SyncAsync();
|
||||||
|
await harness.First.SyncAsync();
|
||||||
|
|
||||||
|
await harness.First.UpdateAsync(entityId, Host("old-db, still needed"));
|
||||||
|
|
||||||
|
await harness.Second.DeleteAsync(entityId);
|
||||||
|
await harness.Second.SyncAsync();
|
||||||
|
harness.Server.CollectTombstones();
|
||||||
|
|
||||||
|
var report = await harness.First.SyncAsync();
|
||||||
|
|
||||||
|
report.Pruned.ShouldBe(0);
|
||||||
|
report.Recreated.ShouldBe(1);
|
||||||
|
report.NeedsAttention.ShouldBeTrue();
|
||||||
|
|
||||||
|
(await harness.First.FindAsync(entityId)).Secret.Label.ShouldBe("old-db, still needed");
|
||||||
|
harness.Server.Find(entityId).ShouldNotBeNull().IsDeleted.ShouldBeFalse();
|
||||||
|
|
||||||
|
(await harness.First.ConflictsAsync())
|
||||||
|
.ShouldContain(conflict => conflict.EntityId == entityId
|
||||||
|
&& conflict.Kind == ConflictKind.RecreatedAfterCollection);
|
||||||
|
|
||||||
|
await harness.Second.SyncAsync();
|
||||||
|
(await harness.Second.FindAsync(entityId)).Secret.Label.ShouldBe("old-db, still needed");
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task AReadSpreadOverSeveralPasses_SweepsAtTheEnd_AndKeepsWhatWasPushedMeanwhile()
|
||||||
|
{
|
||||||
|
// Pages per pass are capped, so the read from the beginning can take more than one pass, and what it
|
||||||
|
// has seen has to outlast each one. A host made here during it is pushed between the passes.
|
||||||
|
using var harness = await CreateAsync(new SyncOptions { PullPageSize = 1, MaxPullPages = 2 });
|
||||||
|
|
||||||
|
harness.Server.MaxPullLimit = 1;
|
||||||
|
|
||||||
|
var gone = await harness.Second.CreateAsync(Host("old-db"));
|
||||||
|
var kept = new List<Guid>();
|
||||||
|
|
||||||
|
for (var index = 0; index < 4; index++)
|
||||||
|
{
|
||||||
|
kept.Add(await harness.Second.CreateAsync(Host($"host-{index}")));
|
||||||
|
}
|
||||||
|
|
||||||
|
// Settled by hand: one pass reads at most two pages, and each needs several.
|
||||||
|
for (var pass = 0; pass < 6; pass++)
|
||||||
|
{
|
||||||
|
await harness.Second.SyncAsync();
|
||||||
|
await harness.First.SyncAsync();
|
||||||
|
}
|
||||||
|
|
||||||
|
await harness.Second.DeleteAsync(gone);
|
||||||
|
await harness.Second.SyncAsync();
|
||||||
|
harness.Server.CollectTombstones();
|
||||||
|
|
||||||
|
kept.Add(await harness.First.CreateAsync(Host("made-meanwhile")));
|
||||||
|
|
||||||
|
var pruned = 0;
|
||||||
|
|
||||||
|
for (var pass = 0; pass < 10; pass++)
|
||||||
|
{
|
||||||
|
pruned += (await harness.First.SyncAsync()).Pruned;
|
||||||
|
|
||||||
|
var state = await harness.First.SyncState.ReadAsync(VaultId, TestContext.Current.CancellationToken);
|
||||||
|
|
||||||
|
if (state.SweepMark is null)
|
||||||
|
{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Mid-sweep, nothing has gone yet.
|
||||||
|
pruned.ShouldBe(0);
|
||||||
|
(await harness.First.ListAsync()).Items.Count.ShouldBe(6);
|
||||||
|
}
|
||||||
|
|
||||||
|
pruned.ShouldBe(1);
|
||||||
|
(await harness.First.ListAsync()).Items.Select(item => item.EntityId).ShouldBe(kept, ignoreOrder: true);
|
||||||
|
harness.Server.Find(kept[^1]).ShouldNotBeNull();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -152,6 +152,36 @@ public sealed class SyncCursorTests
|
|||||||
Should.Throw<ArgumentOutOfRangeException>(() => SyncCursor.Encode(Key, VaultId, -1));
|
Should.Throw<ArgumentOutOfRangeException>(() => SyncCursor.Encode(Key, VaultId, -1));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void CarriesTheHorizonItWasIssuedUnder()
|
||||||
|
{
|
||||||
|
var cursor = SyncCursor.Encode(Key, VaultId, 12, horizon: 40);
|
||||||
|
|
||||||
|
SyncCursor.TryDecode(Key, cursor, VaultId, out var sequence, out var horizon).ShouldBeTrue();
|
||||||
|
sequence.ShouldBe(12);
|
||||||
|
horizon.ShouldBe(40);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void UnderNoHorizon_KeepsTheFirstFormat()
|
||||||
|
{
|
||||||
|
// So a server from before horizons, still running beside a newer one mid-deploy, reads what the newer
|
||||||
|
// one issues for every vault nothing has been collected from.
|
||||||
|
SyncCursor.Encode(Key, VaultId, 12, horizon: 0).ShouldBe(SyncCursor.Encode(Key, VaultId, 12));
|
||||||
|
|
||||||
|
SyncCursor.TryDecode(Key, SyncCursor.Encode(Key, VaultId, 12), VaultId, out _, out var horizon)
|
||||||
|
.ShouldBeTrue();
|
||||||
|
horizon.ShouldBe(0);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void ACursorWithAHorizon_IsStillBoundToItsVault()
|
||||||
|
{
|
||||||
|
var cursor = SyncCursor.Encode(Key, OtherVaultId, 12, horizon: 40);
|
||||||
|
|
||||||
|
SyncCursor.TryDecode(Key, cursor, VaultId, out _, out _).ShouldBeFalse();
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void DifferentSequencesProduceDifferentCursors()
|
public void DifferentSequencesProduceDifferentCursors()
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in new issue
Block a user