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; }