Public Access
Add the encrypted local cache and the sync client
Three new client projects, and the wire-contract fix they needed. DodoSSH.Client.Domain holds the decrypted item model and the three-way merge, with no I/O at all — so the suite that decides whether a credential can be lost runs in milliseconds with nothing to mock. Scalars defer to the server on a genuine clash so every replica resolves the same triple identically and two clients cannot ping-pong; directives merge per name so two people each adding one both keep theirs; the jump chain merges as a whole value because its order is the route. Whatever loses is returned rather than dropped. DodoSSH.Client.Storage is EF Core on SQLite, no SQLCipher: the rows are already ciphertext, so an encrypted file would protect protected bytes at the cost of a native dependency. It keeps the server's state and the outbox in separate tables, which is what preserves the common ancestor a merge needs. One pending operation per item, enforced by a unique index. DodoSSH.Client.Sync is the pull/apply/push loop. Pulling never decrypts — a change with no local work pending is plumbed as ciphertext — so a first sync of thousands of items does not run twice as many AEAD operations for nothing. Contracts: EncryptedPayload gains WrappedDataKey and DataKeyId. The specification has required a per-item data key since crypto.md §3, the columns have existed since the first migration and DshAad.ItemPayload binds the id, but this record had nowhere to put either — so a spec-compliant item could not be transmitted at all. Found by writing the client that has to produce one. Also closes a hole in AadResourceType, which had no value for the HostTag and HostCredential that SyncEntityType has always listed. Four bugs the tests found, not review: - SQLite refuses to order or compare its own DateTimeOffset mapping, and throws at execution rather than model build. Collecting tombstones and listing conflicts are both that shape, so this was a crash waiting for the first user with a deleted host. Timestamps are integers now, by convention so a later field cannot be the one left unconverted. - SQLitePCLRaw 2.1.11, which EF resolves, is covered by GHSA-2m69-gcr7-jv3q. Pinned forward as a family. - Resurrecting content from a remote deletion cleared the original before queueing the copy. Two transactions, so a crash between them lost the work; reversed, and the rescued id is derived from the tombstone so a replay coalesces instead of duplicating. - Several equality assertions went through Shouldly's ShouldBe, which compares IEnumerable element-wise and so tested nothing about the Equals these types exist to provide. Corrected; the falsification that caught it went from 2 failures to 6. The push response's cursor is deliberately ignored. It sits after this client's own writes, so adopting it skips anything another client committed at a lower sequence in the window between a pull and a push — permanently. Re-reading one's own writes is idempotent and costs a page. The Contracts doc that invited the shortcut now says so. 593 tests, up from 448. The delete-versus-edit rules, the ancestor retention, the fresh operation id on coalesce and the cursor safeguard were each verified by breaking them and watching the right test fail.
This commit is contained in:
@@ -0,0 +1,487 @@
|
||||
using System.Globalization;
|
||||
using System.Runtime.InteropServices;
|
||||
using DodoSSH.Client.Api;
|
||||
using DodoSSH.Client.Storage;
|
||||
using DodoSSH.Contracts;
|
||||
|
||||
namespace DodoSSH.Client.Sync;
|
||||
|
||||
/// <summary>
|
||||
/// One vault's synchronisation pass: pull, reconcile, push, pull again.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Pull first, so a local change is merged against the newest server state before it is offered — which
|
||||
/// turns most would-be conflicts into ordinary merges and keeps the push round count down. Push second.
|
||||
/// Pull once more at the end only if something was pushed, so the mirror reflects the versions the
|
||||
/// server actually assigned.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Pulling does not decrypt.</b> A change with no local work pending is copied into the mirror as
|
||||
/// ciphertext and nothing more. Decryption happens when a merge needs it, or when the interface reads an
|
||||
/// item. For a five-thousand-item first sync that is the difference between plumbing bytes and running
|
||||
/// ten thousand AEAD operations for nothing.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class SyncEngine
|
||||
{
|
||||
private readonly ISyncApi api;
|
||||
private readonly ItemStore items;
|
||||
private readonly OutboxStore outbox;
|
||||
private readonly SyncStateStore syncState;
|
||||
private readonly ConflictStore conflicts;
|
||||
private readonly VaultKeyring keyring;
|
||||
private readonly TimeProvider clock;
|
||||
private readonly SyncOptions options;
|
||||
private readonly ItemReconciler reconciler;
|
||||
|
||||
/// <summary>Creates the engine.</summary>
|
||||
public SyncEngine(
|
||||
ISyncApi api,
|
||||
ItemStore items,
|
||||
OutboxStore outbox,
|
||||
SyncStateStore syncState,
|
||||
ConflictStore conflicts,
|
||||
VaultKeyring keyring,
|
||||
TimeProvider clock,
|
||||
SyncOptions? options = null)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(api);
|
||||
ArgumentNullException.ThrowIfNull(items);
|
||||
ArgumentNullException.ThrowIfNull(outbox);
|
||||
ArgumentNullException.ThrowIfNull(syncState);
|
||||
ArgumentNullException.ThrowIfNull(conflicts);
|
||||
ArgumentNullException.ThrowIfNull(keyring);
|
||||
ArgumentNullException.ThrowIfNull(clock);
|
||||
|
||||
this.api = api;
|
||||
this.items = items;
|
||||
this.outbox = outbox;
|
||||
this.syncState = syncState;
|
||||
this.conflicts = conflicts;
|
||||
this.keyring = keyring;
|
||||
this.clock = clock;
|
||||
this.options = options ?? SyncOptions.Default;
|
||||
|
||||
reconciler = new ItemReconciler(items, outbox, conflicts, keyring);
|
||||
}
|
||||
|
||||
/// <summary>Runs a full pass over one vault.</summary>
|
||||
public async Task<SyncReport> SyncAsync(Guid vaultId, CancellationToken cancellationToken)
|
||||
{
|
||||
var report = new SyncReportBuilder(vaultId);
|
||||
|
||||
await PullAsync(vaultId, report, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
var pushedAnything = false;
|
||||
|
||||
for (var round = 1; ; round++)
|
||||
{
|
||||
var outcome = await DrainAsync(vaultId, report, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
if (outcome.Sent == 0)
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
pushedAnything = true;
|
||||
|
||||
if (!outcome.NeedsAnotherRound)
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
if (round >= options.MaxPushRounds)
|
||||
{
|
||||
report.RoundsExhausted = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (pushedAnything)
|
||||
{
|
||||
await PullAsync(vaultId, report, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return report.Build();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Reads every change available and applies it.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The cursor is saved after each page's changes are applied, which makes applying at-least-once
|
||||
/// rather than exactly-once: a process that dies between the two re-reads that page next time. That
|
||||
/// is deliberate and safe, because applying a change is a blind overwrite of a mirror row and a
|
||||
/// resurrection takes a deterministic id. The other ordering — save the cursor first — would lose
|
||||
/// changes outright, which no amount of idempotence can repair.
|
||||
/// </remarks>
|
||||
private async Task PullAsync(
|
||||
Guid vaultId,
|
||||
SyncReportBuilder report,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var state = await syncState.ReadAsync(vaultId, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
for (var page = 0; page < options.MaxPullPages; page++)
|
||||
{
|
||||
var response = await api.SyncPullAsync(
|
||||
vaultId,
|
||||
new SyncPullRequest(state.Cursor, options.PullPageSize, [SyncEntityType.Host]),
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
|
||||
foreach (var change in response.Changes)
|
||||
{
|
||||
await ApplyAsync(vaultId, change, report, cancellationToken).ConfigureAwait(false);
|
||||
report.Pulled++;
|
||||
}
|
||||
|
||||
var advanced = !string.Equals(state.Cursor, response.NextCursor, StringComparison.Ordinal);
|
||||
|
||||
state = Record(vaultId, state, response, report);
|
||||
await syncState.SaveAsync(state, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
if (!response.HasMore)
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
// A server that claims more but neither returns a change nor moves the cursor would spin
|
||||
// this loop for ever. Stopping is the only safe reading of that answer.
|
||||
if (!advanced && response.Changes.Count == 0)
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private StoredSyncState Record(
|
||||
Guid vaultId,
|
||||
StoredSyncState state,
|
||||
SyncPullResponse response,
|
||||
SyncReportBuilder report)
|
||||
{
|
||||
var now = clock.GetUtcNow();
|
||||
var skew = (long)(response.ServerTime - now).TotalMilliseconds;
|
||||
|
||||
report.ServerKeyGeneration = response.CurrentKeyGeneration;
|
||||
report.ServerTimeSkewMs = skew;
|
||||
|
||||
// A generation ahead of the key this client holds means the vault was rekeyed and this client's
|
||||
// grant has not been re-wrapped. Items pulled meanwhile are stored but cannot be read.
|
||||
keyring.TryGet(vaultId, out _, out var held);
|
||||
report.RekeyRequired = response.CurrentKeyGeneration > held;
|
||||
|
||||
return state with
|
||||
{
|
||||
Cursor = response.NextCursor,
|
||||
KeyGeneration = response.CurrentKeyGeneration,
|
||||
LastPulledAt = now,
|
||||
ServerTimeSkewMs = skew,
|
||||
};
|
||||
}
|
||||
|
||||
private async Task ApplyAsync(
|
||||
Guid vaultId,
|
||||
SyncChange change,
|
||||
SyncReportBuilder report,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (change.EntityType != SyncEntityType.Host)
|
||||
{
|
||||
// Reserved in the contract but not yet syncable. Ignoring it keeps a newer server's extra
|
||||
// entity types from breaking an older client's pull.
|
||||
return;
|
||||
}
|
||||
|
||||
await reconciler.MirrorAsync(vaultId, change, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
var pending = await outbox
|
||||
.FindAsync(vaultId, change.EntityType, change.EntityId, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
if (pending is null || pending.IsParked)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await reconciler.ReconcileAsync(vaultId, change, pending, report, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>What one push round achieved.</summary>
|
||||
[StructLayout(LayoutKind.Auto)]
|
||||
private readonly record struct DrainOutcome(int Sent, int Conflicts, bool BatchWasFull)
|
||||
{
|
||||
internal bool NeedsAnotherRound => Conflicts > 0 || BatchWasFull;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends one batch and acts on each per-operation answer.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <b>The cursor in the push response is deliberately ignored.</b> It sits after this push's own
|
||||
/// changes, so adopting it would skip any change another client committed at a lower sequence
|
||||
/// between this client's last pull and this push — permanently. Continuing from the cursor this
|
||||
/// client already holds re-reads its own writes, which costs one redundant page and is idempotent.
|
||||
/// </remarks>
|
||||
private async Task<DrainOutcome> DrainAsync(
|
||||
Guid vaultId,
|
||||
SyncReportBuilder report,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var pending = await outbox
|
||||
.TakeAsync(vaultId, options.MaxOperationsPerPush, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
if (pending.Count == 0)
|
||||
{
|
||||
return default;
|
||||
}
|
||||
|
||||
var operations = new List<SyncPushOperation>(pending.Count);
|
||||
var byOperationId = new Dictionary<Guid, PendingOperation>(pending.Count);
|
||||
|
||||
foreach (var operation in pending)
|
||||
{
|
||||
operations.Add(new SyncPushOperation(
|
||||
operation.OperationId,
|
||||
operation.EntityType,
|
||||
operation.EntityId,
|
||||
operation.Operation,
|
||||
operation.ExpectedVersion,
|
||||
operation.Payload,
|
||||
operation.Fields));
|
||||
|
||||
byOperationId[operation.OperationId] = operation;
|
||||
|
||||
await outbox.MarkDispatchedAsync(operation.Sequence, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
var response = await api
|
||||
.SyncPushAsync(vaultId, new SyncPushRequest(operations), cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
var conflicted = 0;
|
||||
|
||||
foreach (var result in response.Results)
|
||||
{
|
||||
if (!byOperationId.TryGetValue(result.OperationId, out var operation))
|
||||
{
|
||||
// An id this client did not send. Nothing sane to do with it.
|
||||
continue;
|
||||
}
|
||||
|
||||
// The attempt count was incremented above, so the bound is read from the fresh value.
|
||||
var attempts = operation.Attempts + 1;
|
||||
|
||||
if (await HandleAsync(vaultId, operation with { Attempts = attempts }, result, report, cancellationToken)
|
||||
.ConfigureAwait(false))
|
||||
{
|
||||
conflicted++;
|
||||
}
|
||||
}
|
||||
|
||||
return new DrainOutcome(
|
||||
pending.Count, conflicted, pending.Count == options.MaxOperationsPerPush);
|
||||
}
|
||||
|
||||
/// <returns>Whether this answer warrants another push round.</returns>
|
||||
private async Task<bool> HandleAsync(
|
||||
Guid vaultId,
|
||||
PendingOperation operation,
|
||||
SyncPushResult result,
|
||||
SyncReportBuilder report,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
switch (result.Status)
|
||||
{
|
||||
case SyncOperationStatus.Applied:
|
||||
case SyncOperationStatus.Duplicate:
|
||||
// Duplicate means an earlier push of this exact operation id already landed, so the
|
||||
// stored state is what this operation intended. Treated as success on purpose: that is
|
||||
// what makes a retry after a timeout exactly-once rather than merely at-least-once.
|
||||
await AcceptAsync(vaultId, operation, result, cancellationToken).ConfigureAwait(false);
|
||||
report.Pushed++;
|
||||
return false;
|
||||
|
||||
case SyncOperationStatus.Conflict:
|
||||
return await ResolveAsync(vaultId, operation, result, report, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
case SyncOperationStatus.Forbidden:
|
||||
await RejectAsync(
|
||||
vaultId,
|
||||
operation,
|
||||
"You no longer have permission to change this item.",
|
||||
report,
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
return false;
|
||||
|
||||
case SyncOperationStatus.Invalid:
|
||||
await RejectAsync(
|
||||
vaultId,
|
||||
operation,
|
||||
result.Detail ?? "The server rejected this change as invalid.",
|
||||
report,
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
return false;
|
||||
|
||||
default:
|
||||
await outbox.RecordFailureAsync(
|
||||
operation.Sequence,
|
||||
$"Unexpected push status {result.Status}.",
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Records an accepted operation and clears it from the outbox.</summary>
|
||||
private async Task AcceptAsync(
|
||||
Guid vaultId,
|
||||
PendingOperation operation,
|
||||
SyncPushResult result,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var expected = SyncVersions.NextVersion(operation.ExpectedVersion);
|
||||
var version = result.Version ?? expected;
|
||||
var isDelete = operation.Operation == SyncOperation.Delete;
|
||||
|
||||
// The payload was sealed at the version this client predicted, and the AAD binds that version.
|
||||
// If the server assigned a different one — which its own version check should make impossible —
|
||||
// storing the payload would leave a mirror row that never decrypts. Skip the write; the pull at
|
||||
// the end of the pass brings the authoritative row.
|
||||
var canMirror = isDelete || version == expected;
|
||||
|
||||
if (canMirror)
|
||||
{
|
||||
await items.SaveAsync(
|
||||
new StoredItem(
|
||||
vaultId,
|
||||
operation.EntityType,
|
||||
operation.EntityId,
|
||||
version,
|
||||
result.ChangeSequence ?? 0,
|
||||
isDelete ? null : operation.Payload,
|
||||
isDelete ? null : operation.Fields,
|
||||
isDelete,
|
||||
clock.GetUtcNow()),
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
await outbox.CompleteAsync(operation.Sequence, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task<bool> ResolveAsync(
|
||||
Guid vaultId,
|
||||
PendingOperation operation,
|
||||
SyncPushResult result,
|
||||
SyncReportBuilder report,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (operation.Attempts >= options.MaxAttemptsBeforeParking)
|
||||
{
|
||||
await RejectAsync(
|
||||
vaultId,
|
||||
operation,
|
||||
string.Create(
|
||||
CultureInfo.InvariantCulture,
|
||||
$"Could not be reconciled after {operation.Attempts} attempts."),
|
||||
report,
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
if (result.ServerEntity is null)
|
||||
{
|
||||
// The version check failed but the server has no such row. Re-offer it as a create.
|
||||
return await RetryAsCreateAsync(vaultId, operation, report, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
await reconciler.MirrorAsync(vaultId, result.ServerEntity, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
await reconciler
|
||||
.ReconcileAsync(vaultId, result.ServerEntity, operation, report, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
private async Task<bool> RetryAsCreateAsync(
|
||||
Guid vaultId,
|
||||
PendingOperation operation,
|
||||
SyncReportBuilder report,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (operation.Operation == SyncOperation.Delete)
|
||||
{
|
||||
// Nothing there to delete, so the intent is already satisfied.
|
||||
await outbox.CompleteAsync(operation.Sequence, cancellationToken).ConfigureAwait(false);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!keyring.TryGet(vaultId, out var vaultKey, out var generation)
|
||||
|| operation.Payload is null)
|
||||
{
|
||||
await RejectAsync(
|
||||
vaultId, operation, "This item has no usable vault key.", report, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
return false;
|
||||
}
|
||||
|
||||
var local = HostCipher.TryOpen(
|
||||
operation.Payload,
|
||||
vaultKey.Span,
|
||||
operation.EntityId,
|
||||
SyncVersions.NextVersion(operation.ExpectedVersion));
|
||||
|
||||
if (local is null)
|
||||
{
|
||||
await RejectAsync(
|
||||
vaultId,
|
||||
operation,
|
||||
"The queued change could not be decrypted, so it could not be re-offered.",
|
||||
report,
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Re-sealed at version 1, because that is what the server assigns to a create and the AAD binds
|
||||
// the version.
|
||||
await outbox.ReviseAsync(
|
||||
operation.Sequence,
|
||||
SyncOperation.Upsert,
|
||||
expectedVersion: null,
|
||||
HostCipher.Seal(local.Host, vaultKey.Span, operation.EntityId, generation, itemVersion: 1),
|
||||
HostFields.From(local.Host),
|
||||
ancestor: null,
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>Parks an operation the server will never accept, and says why.</summary>
|
||||
private async Task RejectAsync(
|
||||
Guid vaultId,
|
||||
PendingOperation operation,
|
||||
string reason,
|
||||
SyncReportBuilder report,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
await outbox.ParkAsync(operation.Sequence, reason, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
await conflicts.RecordAsync(
|
||||
vaultId,
|
||||
operation.EntityType,
|
||||
operation.EntityId,
|
||||
ConflictKind.Rejected,
|
||||
ConflictDetailCodec.Encode(reason),
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
|
||||
report.Parked++;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user