using DodoSSH.Contracts;
using static DodoSSH.Client.Storage.Tests.CacheHarness;
namespace DodoSSH.Client.Storage.Tests;
///
/// The outbox, whose coalescing rules are where offline work is kept or lost.
///
///
/// 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 Duplicate for an operation whose contents have since changed and silently discard
/// the newer edit.
///
public sealed class OutboxStoreTests : IAsyncLifetime
{
private CacheHarness harness = null!;
///
public async ValueTask InitializeAsync() => harness = await CreateAsync();
///
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.Sequence,
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.Sequence, "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.Sequence, "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.Sequence, 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.Sequence, Token)).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(
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;
}