Public Access
jaap-jan
f332d9a56f
- The outbox coalesces a new save into the row a push already read, under a new operation id, but Complete, Revise and Park found the row by sequence alone. The acknowledgement of the earlier state deleted the row holding the later one and the edit was gone without a word. They now take the PendingOperation and act only while the row still holds it; a later edit is rebased onto the version just accepted, so it pushes cleanly instead of conflicting with its own earlier save. - A pull filtered by type moves the cursor past changes of types it did not ask for, so a build that added a type (SpelunkKey, ObjectStore) never received the items of that type made before it. sync_state.pulled_types records the list a cursor was read under, and a different list starts the pull over once. Migration AddSyncStatePulledTypes.
318 lines
13 KiB
C#
318 lines
13 KiB
C#
using DodoSSH.Contracts;
|
|
using static DodoSSH.Client.Storage.Tests.CacheHarness;
|
|
|
|
namespace DodoSSH.Client.Storage.Tests;
|
|
|
|
/// <summary>
|
|
/// The outbox, whose coalescing rules are where offline work is kept or lost.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Two properties carry the weight. The ancestor must survive every coalesce, or a conflict can only
|
|
/// be arbitrated rather than merged. And a coalesced row must get a fresh operation id, or the server
|
|
/// can answer <c>Duplicate</c> for an operation whose contents have since changed and silently discard
|
|
/// the newer edit.
|
|
/// </remarks>
|
|
public sealed class OutboxStoreTests : IAsyncLifetime
|
|
{
|
|
private CacheHarness harness = null!;
|
|
|
|
/// <inheritdoc />
|
|
public async ValueTask InitializeAsync() => harness = await CreateAsync();
|
|
|
|
/// <inheritdoc />
|
|
public ValueTask DisposeAsync()
|
|
{
|
|
harness.Dispose();
|
|
return ValueTask.CompletedTask;
|
|
}
|
|
|
|
[Fact]
|
|
public async Task AQueuedChange_ComesBackWithEverythingItNeedsToBePushed()
|
|
{
|
|
var entityId = Guid.CreateVersion7();
|
|
var ancestor = new StoredAncestor(3, Payload(seed: 40), new SyncPlaintextFields());
|
|
|
|
var queued = await harness.Outbox.QueueAsync(
|
|
Change(entityId, expectedVersion: 3, seed: 7, ancestor: ancestor), Token);
|
|
|
|
queued.OperationId.ShouldNotBe(Guid.Empty);
|
|
queued.ExpectedVersion.ShouldBe(3);
|
|
queued.Operation.ShouldBe(SyncOperation.Upsert);
|
|
queued.Payload.ShouldNotBeNull();
|
|
queued.Payload.Envelope.ShouldBe(Payload(seed: 7).Envelope);
|
|
queued.Payload.WrappedDataKey.ShouldBe(Payload(seed: 7).WrappedDataKey);
|
|
queued.Payload.DataKeyId.ShouldBe(Payload(seed: 7).DataKeyId);
|
|
queued.Ancestor.ShouldNotBeNull();
|
|
queued.Ancestor.Version.ShouldBe(3);
|
|
queued.Ancestor.Payload.Envelope.ShouldBe(Payload(seed: 40).Envelope);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ASecondEditToTheSameItem_CoalescesIntoOneRow()
|
|
{
|
|
// Two rows would have to be pushed in order, and the second's expectedVersion is the version
|
|
// the first will produce — which is not known when it is queued.
|
|
var entityId = Guid.CreateVersion7();
|
|
|
|
await harness.Outbox.QueueAsync(Change(entityId, seed: 1), Token);
|
|
await harness.Outbox.QueueAsync(Change(entityId, seed: 2), Token);
|
|
|
|
var pending = await harness.Outbox.TakeAsync(VaultId, 10, Token);
|
|
|
|
pending.ShouldHaveSingleItem().Payload!.Envelope.ShouldBe(Payload(seed: 2).Envelope);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ACoalescedEdit_KeepsTheOriginalAncestorAndExpectedVersion()
|
|
{
|
|
// The load-bearing rule. The newest state is still a descendant of the base the first edit
|
|
// branched from; adopting the caller's values here would discard the common ancestor after the
|
|
// first edit and leave nothing to merge against.
|
|
var entityId = Guid.CreateVersion7();
|
|
var ancestor = new StoredAncestor(5, Payload(seed: 90), new SyncPlaintextFields());
|
|
|
|
await harness.Outbox.QueueAsync(
|
|
Change(entityId, expectedVersion: 5, seed: 1, ancestor: ancestor), Token);
|
|
|
|
// A second edit arrives knowing nothing about the base.
|
|
await harness.Outbox.QueueAsync(
|
|
Change(entityId, expectedVersion: null, seed: 2, ancestor: null), Token);
|
|
|
|
var pending = (await harness.Outbox.TakeAsync(VaultId, 10, Token)).ShouldHaveSingleItem();
|
|
|
|
pending.ExpectedVersion.ShouldBe(5);
|
|
pending.Ancestor.ShouldNotBeNull();
|
|
pending.Ancestor.Version.ShouldBe(5);
|
|
pending.Ancestor.Payload.Envelope.ShouldBe(Payload(seed: 90).Envelope);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ACoalescedEdit_GetsAFreshOperationId()
|
|
{
|
|
// Reusing the id would let the server report Duplicate — meaning "already applied" — for an
|
|
// operation whose payload has since changed, and the newer edit would vanish with the push
|
|
// reported as a success.
|
|
var entityId = Guid.CreateVersion7();
|
|
|
|
var first = await harness.Outbox.QueueAsync(Change(entityId, seed: 1), Token);
|
|
await harness.Outbox.MarkDispatchedAsync(first.Sequence, Token);
|
|
|
|
var second = await harness.Outbox.QueueAsync(Change(entityId, seed: 2), Token);
|
|
|
|
second.OperationId.ShouldNotBe(first.OperationId);
|
|
second.Sequence.ShouldBe(first.Sequence);
|
|
|
|
// And the retry counter resets, because this is a new operation rather than a further attempt
|
|
// at the old one.
|
|
second.Attempts.ShouldBe(0);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task AnUpsertFollowedByADelete_BecomesADelete()
|
|
{
|
|
var entityId = Guid.CreateVersion7();
|
|
|
|
await harness.Outbox.QueueAsync(Change(entityId, expectedVersion: 2, seed: 1), Token);
|
|
await harness.Outbox.QueueAsync(
|
|
Change(entityId, SyncOperation.Delete, expectedVersion: 2), Token);
|
|
|
|
var pending = (await harness.Outbox.TakeAsync(VaultId, 10, Token)).ShouldHaveSingleItem();
|
|
|
|
pending.Operation.ShouldBe(SyncOperation.Delete);
|
|
pending.Payload.ShouldBeNull();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ChangesToDifferentItems_DrainInTheOrderTheyWereMade()
|
|
{
|
|
// Order matters for creates that reference each other — a host naming a jump host — so the
|
|
// outbox is a queue, not a set.
|
|
var first = Guid.CreateVersion7();
|
|
var second = Guid.CreateVersion7();
|
|
var third = Guid.CreateVersion7();
|
|
|
|
foreach (var id in new[] { first, second, third })
|
|
{
|
|
await harness.Outbox.QueueAsync(Change(id), Token);
|
|
}
|
|
|
|
var pending = await harness.Outbox.TakeAsync(VaultId, 10, Token);
|
|
|
|
pending.Select(p => p.EntityId).ShouldBe([first, second, third]);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task CoalescingDoesNotJumpTheQueue()
|
|
{
|
|
// The row keeps its original position. Re-editing the first item should not push it behind
|
|
// items queued after it, because the later ones may depend on it existing.
|
|
var first = Guid.CreateVersion7();
|
|
var second = Guid.CreateVersion7();
|
|
|
|
await harness.Outbox.QueueAsync(Change(first), Token);
|
|
await harness.Outbox.QueueAsync(Change(second), Token);
|
|
await harness.Outbox.QueueAsync(Change(first, seed: 9), Token);
|
|
|
|
var pending = await harness.Outbox.TakeAsync(VaultId, 10, Token);
|
|
|
|
pending.Select(p => p.EntityId).ShouldBe([first, second]);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Revise_MovesTheAncestorForwardUnlikeQueue()
|
|
{
|
|
// The opposite intent from a coalesce: a merge has just been performed against a newer server
|
|
// version, so that version becomes the base. Leaving the old ancestor would make the re-push
|
|
// conflict against the same point for ever.
|
|
var entityId = Guid.CreateVersion7();
|
|
var original = new StoredAncestor(1, Payload(seed: 10), new SyncPlaintextFields());
|
|
|
|
var queued = await harness.Outbox.QueueAsync(
|
|
Change(entityId, expectedVersion: 1, ancestor: original), Token);
|
|
|
|
var merged = new StoredAncestor(4, Payload(seed: 20), new SyncPlaintextFields());
|
|
|
|
var revised = await harness.Outbox.ReviseAsync(
|
|
queued,
|
|
SyncOperation.Upsert,
|
|
expectedVersion: 4,
|
|
Payload(seed: 30),
|
|
new SyncPlaintextFields(),
|
|
merged,
|
|
Token);
|
|
|
|
revised.ShouldNotBeNull();
|
|
revised.ExpectedVersion.ShouldBe(4);
|
|
revised.Ancestor!.Version.ShouldBe(4);
|
|
revised.Ancestor.Payload.Envelope.ShouldBe(Payload(seed: 20).Envelope);
|
|
revised.OperationId.ShouldNotBe(queued.OperationId);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task AParkedOperation_IsNotHandedOutForPushing()
|
|
{
|
|
// An operation the server called Invalid will never succeed. Retrying it would spin and, worse,
|
|
// would block every change queued behind it in a vault the user can still write to.
|
|
var parked = Guid.CreateVersion7();
|
|
var healthy = Guid.CreateVersion7();
|
|
|
|
var queued = await harness.Outbox.QueueAsync(Change(parked), Token);
|
|
await harness.Outbox.QueueAsync(Change(healthy), Token);
|
|
|
|
await harness.Outbox.ParkAsync(queued, "Entity type not supported.", Token);
|
|
|
|
var pending = await harness.Outbox.TakeAsync(VaultId, 10, Token);
|
|
pending.ShouldHaveSingleItem().EntityId.ShouldBe(healthy);
|
|
|
|
var listed = (await harness.Outbox.ListParkedAsync(VaultId, Token)).ShouldHaveSingleItem();
|
|
listed.EntityId.ShouldBe(parked);
|
|
listed.LastError.ShouldBe("Entity type not supported.");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ReEditingAParkedOperation_Unparks()
|
|
{
|
|
// The user's remedy for a rejected change is to change it. That has to actually re-arm it.
|
|
var entityId = Guid.CreateVersion7();
|
|
|
|
var queued = await harness.Outbox.QueueAsync(Change(entityId), Token);
|
|
await harness.Outbox.ParkAsync(queued, "nope", Token);
|
|
|
|
var requeued = await harness.Outbox.QueueAsync(Change(entityId, seed: 5), Token);
|
|
|
|
requeued.IsParked.ShouldBeFalse();
|
|
requeued.LastError.ShouldBeNull();
|
|
(await harness.Outbox.TakeAsync(VaultId, 10, Token)).ShouldHaveSingleItem();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Complete_RemovesTheOperation()
|
|
{
|
|
var queued = await harness.Outbox.QueueAsync(Change(Guid.CreateVersion7()), Token);
|
|
|
|
(await harness.Outbox.CompleteAsync(queued, Token)).ShouldBeTrue();
|
|
(await harness.Outbox.TakeAsync(VaultId, 10, Token)).ShouldBeEmpty();
|
|
|
|
// Idempotent: a drain that retries after a crash must not fail on an already-cleared row.
|
|
(await harness.Outbox.CompleteAsync(queued, Token)).ShouldBeFalse();
|
|
}
|
|
|
|
/// <summary>
|
|
/// An answer to an operation that has since been replaced by a newer edit applies to nothing.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// The newer edit coalesced into the same row, so the sequence still matches; only the operation id says
|
|
/// the row is no longer the one that was sent. Completing, revising or parking it would act on the newer
|
|
/// edit on the strength of an answer about the older one.
|
|
/// </remarks>
|
|
[Fact]
|
|
public async Task AnAnswerToAReplacedOperation_LeavesTheNewerEditAlone()
|
|
{
|
|
var entityId = Guid.CreateVersion7();
|
|
|
|
var sent = await harness.Outbox.QueueAsync(Change(entityId, seed: 1), Token);
|
|
var newer = await harness.Outbox.QueueAsync(Change(entityId, seed: 2), Token);
|
|
|
|
newer.Sequence.ShouldBe(sent.Sequence, "the newer edit coalesces into the same row");
|
|
|
|
(await harness.Outbox.CompleteAsync(sent, Token)).ShouldBeFalse();
|
|
await harness.Outbox.ParkAsync(sent, "refused", Token);
|
|
|
|
(await harness.Outbox.ReviseAsync(
|
|
sent, sent.Operation, sent.ExpectedVersion, sent.Payload, sent.Fields, sent.Ancestor, Token))
|
|
.ShouldBeNull();
|
|
|
|
var left = (await harness.Outbox.TakeAsync(VaultId, 10, Token)).ShouldHaveSingleItem();
|
|
|
|
left.OperationId.ShouldBe(newer.OperationId);
|
|
left.IsParked.ShouldBeFalse();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task MarkDispatched_CountsAttempts()
|
|
{
|
|
var queued = await harness.Outbox.QueueAsync(Change(Guid.CreateVersion7()), Token);
|
|
|
|
await harness.Outbox.MarkDispatchedAsync(queued.Sequence, Token);
|
|
await harness.Outbox.MarkDispatchedAsync(queued.Sequence, Token);
|
|
|
|
var pending = (await harness.Outbox.TakeAsync(VaultId, 10, Token)).ShouldHaveSingleItem();
|
|
pending.Attempts.ShouldBe(2);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task AnUpsertWithoutAPayload_IsRefused()
|
|
{
|
|
// Caught here rather than at the server, where it would come back as one opaque Invalid among
|
|
// a batch of otherwise good operations.
|
|
var change = new QueuedChange(
|
|
VaultId,
|
|
SyncEntityType.Host,
|
|
Guid.CreateVersion7(),
|
|
SyncOperation.Upsert,
|
|
ExpectedVersion: null,
|
|
Payload: null,
|
|
Fields: null,
|
|
Ancestor: null);
|
|
|
|
await Should.ThrowAsync<ArgumentException>(
|
|
async () => await harness.Outbox.QueueAsync(change, Token));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ThePendingOperationForAnItem_CanBeLookedUpDirectly()
|
|
{
|
|
// How a pull discovers that an incoming change collides with local work.
|
|
var entityId = Guid.CreateVersion7();
|
|
await harness.Outbox.QueueAsync(Change(entityId), Token);
|
|
|
|
(await harness.Outbox.FindAsync(VaultId, SyncEntityType.Host, entityId, Token))
|
|
.ShouldNotBeNull();
|
|
|
|
(await harness.Outbox.FindAsync(VaultId, SyncEntityType.Host, Guid.CreateVersion7(), Token))
|
|
.ShouldBeNull();
|
|
}
|
|
|
|
private static CancellationToken Token => TestContext.Current.CancellationToken;
|
|
}
|