532 lines
31 KiB
C#
532 lines
31 KiB
C#
using System.Net;
|
|
using System.Net.Http;
|
|
using System.Text;
|
|
using game.Account;
|
|
using game.PlayerDataStorage;
|
|
|
|
static class Test
|
|
{
|
|
static int passed, failed;
|
|
static readonly TimeSpan Limit = TimeSpan.FromSeconds(5);
|
|
|
|
static async Task Main()
|
|
{
|
|
await Run("create-empty-and-fixed-cookie", CreateAndFixedCookie);
|
|
await Run("edit-flush-confirm", EditFlushConfirm);
|
|
await Run("superseded-before-send", SupersededBeforeSend);
|
|
await Run("local-only-no-ticket", LocalOnlyNoTicket);
|
|
await Run("wait-cancel-does-not-cancel-flush", WaitCancellation);
|
|
await Run("http500-becomes-unknown", HttpFailureUnknown);
|
|
await Run("two-owner-cookie-isolation", CookieIsolation);
|
|
await Run("create-validation-and-unauthenticated", CreateValidation);
|
|
await Run("create-get401-releases-cache-lock", CreateFailureReleasesLock);
|
|
await Run("busy-until-owner-close", BusyUntilClose);
|
|
await Run("set-delete-order-and-batch10", SetDeleteAndBatching);
|
|
await Run("fifty-one-keys-are-drained", FiftyOneKeys);
|
|
await Run("flush-watermark-leaves-new-version", WatermarkAndInFlightVersion);
|
|
await Run("flush-wait-cancel-does-not-cancel-http", FlushWaitCancellation);
|
|
await Run("network-failure-becomes-unknown", NetworkFailureUnknown);
|
|
await Run("close-is-idempotent", CloseIdempotent);
|
|
await Run("close-waits-for-ignoring-transport", CloseWaitsForTransport);
|
|
await Run("stale-epoch-blocks-new-edits", StaleEpoch);
|
|
await Run("active-write-disk-failure-does-not-send", ActiveWriteDiskFailure);
|
|
await Run("http200-confirm-disk-failure-is-unknown", ConfirmDiskFailure);
|
|
await Run("unauthorized-write-is-unknown-and-owned", UnauthorizedWrite);
|
|
await Run("queued-flush-close-returns-structured-result", QueuedFlushClose);
|
|
await Run("queued-refresh-close-returns-structured-result", QueuedRefreshClose);
|
|
await Run("create-protocol-network-cancel-errors", CreateReadErrors);
|
|
await Run("corrupt-cache-requires-recovery", RecoveryRequired);
|
|
await Run("disposed-login-session-invalidates-owner", LoginSessionInvalidation);
|
|
await Run("http400-403-and-timeout-are-unknown", MutationFailureVariants);
|
|
await Run("close-cancel-callback-reentry-does-not-deadlock", CancelCallbackReentry);
|
|
await Run("recovered-pending-is-counted-by-flush", RecoveredPendingCount);
|
|
Console.WriteLine($"RESULT passed={passed} failed={failed}");
|
|
Environment.ExitCode = failed == 0 ? 0 : 1;
|
|
}
|
|
|
|
static async Task Run(string name, Func<Task> body)
|
|
{
|
|
try { await body().WaitAsync(Limit); passed++; Console.WriteLine("PASS " + name); }
|
|
catch (Exception ex) { failed++; Console.WriteLine("FAIL " + name + " " + ex.GetType().Name + ": " + ex.Message); }
|
|
}
|
|
|
|
static async Task CreateAndFixedCookie()
|
|
{
|
|
var seenJar = default(CookieContainer);
|
|
var session = await Login("101");
|
|
var root = Root();
|
|
var result = await PlayerKVSessionOwner.CreateAsync(session, root, "101", new Uri("http://unit.test/"), 7,
|
|
e => e == 7, default, jar => { seenJar = jar; return new JsonHandler(_ => OkGet()); });
|
|
Check(result.Success, "owner create");
|
|
Check(ReferenceEquals(seenJar, session.Cookies), "captured cookie jar");
|
|
await result.Owner.CloseAsync();
|
|
session.Dispose();
|
|
}
|
|
|
|
static async Task EditFlushConfirm()
|
|
{
|
|
var handler = new JsonHandler(_ => OkMutationOrGet(_));
|
|
var (session, owner) = await Owner("101", handler);
|
|
var edit = owner.Update("a", "v");
|
|
Check(edit.Accepted && edit.Ticket != null, "accepted ticket");
|
|
var flush = await owner.FlushAsync();
|
|
Check(flush.Confirmed == 1 && flush.Unknown == 0, "flush confirmed");
|
|
Check((await owner.WaitForTicketAsync(edit.Ticket)).Status == PlayerKVTicketStatus.Confirmed, "ticket confirmed");
|
|
Check(handler.MaxInFlight == 1, "single transport");
|
|
await owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task SupersededBeforeSend()
|
|
{
|
|
var (session, owner) = await Owner("101", new JsonHandler(_ => OkMutationOrGet(_)));
|
|
var first = owner.Update("same", "v1").Ticket;
|
|
var second = owner.Update("same", "v2").Ticket;
|
|
Check((await owner.WaitForTicketAsync(first)).Status == PlayerKVTicketStatus.Superseded, "old superseded");
|
|
await owner.FlushAsync();
|
|
Check((await owner.WaitForTicketAsync(second)).Status == PlayerKVTicketStatus.Confirmed, "new confirmed");
|
|
await owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task LocalOnlyNoTicket()
|
|
{
|
|
var (session, owner) = await Owner("101", new JsonHandler(_ => OkMutationOrGet(_)));
|
|
var result = owner.Update("local", "v", false);
|
|
Check(result.Accepted && result.Ticket == null, "local result");
|
|
var flush = await owner.FlushAsync();
|
|
Check(flush.Confirmed == 0, "no remote ticket");
|
|
await owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task WaitCancellation()
|
|
{
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var handler = new JsonHandler(async request =>
|
|
{
|
|
if (request.RequestUri.AbsolutePath.EndsWith("/Get")) return OkGet();
|
|
entered.TrySetResult(true); await release.Task; return Ok();
|
|
});
|
|
var (session, owner) = await Owner("101", handler);
|
|
var ticket = owner.Update("a", "v").Ticket;
|
|
var flushTask = owner.FlushAsync();
|
|
await entered.Task;
|
|
using var cts = new CancellationTokenSource(); cts.Cancel();
|
|
Check((await owner.WaitForTicketAsync(ticket, cts.Token)).Status == PlayerKVTicketStatus.Cancelled, "wait cancelled");
|
|
release.TrySetResult(true);
|
|
await flushTask;
|
|
Check((await owner.WaitForTicketAsync(ticket)).Status == PlayerKVTicketStatus.Confirmed, "fact continues");
|
|
await owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task HttpFailureUnknown()
|
|
{
|
|
var handler = new JsonHandler(request => Task.FromResult(request.RequestUri.AbsolutePath.EndsWith("/Get")
|
|
? OkGet() : new HttpResponseMessage(HttpStatusCode.InternalServerError)));
|
|
var (session, owner) = await Owner("101", handler);
|
|
var ticket = owner.Update("a", "v").Ticket;
|
|
var flush = await owner.FlushAsync();
|
|
Check(flush.Unknown == 1, "unknown count");
|
|
Check((await owner.WaitForTicketAsync(ticket)).Status == PlayerKVTicketStatus.Unknown, "ticket unknown");
|
|
await owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task CookieIsolation()
|
|
{
|
|
var s1 = await Login("101"); var s2 = await Login("102");
|
|
CookieContainer j1 = null, j2 = null;
|
|
var r1 = await PlayerKVSessionOwner.CreateAsync(s1, Root(), "101", new Uri("http://unit.test/"), 1, e => e == 1,
|
|
default, jar => { j1 = jar; return new JsonHandler(_ => OkGet()); });
|
|
var r2 = await PlayerKVSessionOwner.CreateAsync(s2, Root(), "102", new Uri("http://unit.test/"), 2, e => e == 2,
|
|
default, jar => { j2 = jar; return new JsonHandler(_ => OkGet()); });
|
|
Check(r1.Success && r2.Success && !ReferenceEquals(j1, j2), "isolated jars");
|
|
await r1.Owner.CloseAsync(); await r2.Owner.CloseAsync(); s1.Dispose(); s2.Dispose();
|
|
}
|
|
|
|
static async Task CreateValidation()
|
|
{
|
|
var plain = new S101AccountSession(new Uri("http://unit.test/"));
|
|
var unauth = await PlayerKVSessionOwner.CreateAsync(plain, Root(), "101", new Uri("http://unit.test/"), 1, _ => true);
|
|
Check(unauth.Error == PlayerKVSessionOpenError.Unauthenticated, "unauthenticated");
|
|
plain.Dispose();
|
|
var session = await Login("101");
|
|
var wrongId = await PlayerKVSessionOwner.CreateAsync(session, Root(), "102", new Uri("http://unit.test/"), 1, _ => true);
|
|
var wrongApi = await PlayerKVSessionOwner.CreateAsync(session, Root(), "101", new Uri("http://other.test/"), 1, _ => true);
|
|
Check(wrongId.Error == PlayerKVSessionOpenError.Stale && wrongApi.Error == PlayerKVSessionOpenError.Stale, "identity mismatch");
|
|
session.Dispose();
|
|
}
|
|
|
|
static async Task CreateFailureReleasesLock()
|
|
{
|
|
var root = Root(); var session = await Login("101");
|
|
var failed = await PlayerKVSessionOwner.CreateAsync(session, root, "101", new Uri("http://unit.test/"), 1, _ => true,
|
|
default, _ => new JsonHandler(_ => new HttpResponseMessage(HttpStatusCode.Unauthorized)));
|
|
Check(failed.Error == PlayerKVSessionOpenError.Unauthorized, "401 create");
|
|
var retry = await PlayerKVSessionOwner.CreateAsync(session, root, "101", new Uri("http://unit.test/"), 1, _ => true,
|
|
default, _ => new JsonHandler(_ => OkGet()));
|
|
Check(retry.Success, "lock released");
|
|
await retry.Owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task BusyUntilClose()
|
|
{
|
|
var root = Root(); var session = await Login("101");
|
|
var first = await PlayerKVSessionOwner.CreateAsync(session, root, "101", new Uri("http://unit.test/"), 1, _ => true,
|
|
default, _ => new JsonHandler(_ => OkGet()));
|
|
var busy = await PlayerKVSessionOwner.CreateAsync(session, root, "101", new Uri("http://unit.test/"), 1, _ => true,
|
|
default, _ => new JsonHandler(_ => OkGet()));
|
|
Check(busy.Error == PlayerKVSessionOpenError.Busy, "busy");
|
|
await first.Owner.CloseAsync();
|
|
var reopened = await PlayerKVSessionOwner.CreateAsync(session, root, "101", new Uri("http://unit.test/"), 2, _ => true,
|
|
default, _ => new JsonHandler(_ => OkGet()));
|
|
Check(reopened.Success, "reopen after close"); await reopened.Owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task SetDeleteAndBatching()
|
|
{
|
|
var paths = new List<string>();
|
|
var handler = new JsonHandler(request => { lock (paths) paths.Add(request.RequestUri.AbsolutePath); return OkMutationOrGet(request); });
|
|
var (session, owner) = await Owner("101", handler);
|
|
var tickets = new List<PlayerKVSyncTicket>();
|
|
for (var i = 0; i < 12; i++) tickets.Add(owner.Update("s" + i, "v").Ticket);
|
|
tickets.Add(owner.Delete("gone").Ticket);
|
|
await owner.FlushAsync();
|
|
Check(paths.Count(p => p.EndsWith("/Set")) == 2 && paths.Count(p => p.EndsWith("/Delete")) == 1, "10+2+delete batches");
|
|
Check(tickets.All(t => owner.WaitForTicketAsync(t).GetAwaiter().GetResult().Status == PlayerKVTicketStatus.Confirmed), "all confirmed");
|
|
await owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task FiftyOneKeys()
|
|
{
|
|
var mutations = 0;
|
|
var handler = new JsonHandler(request => { if (!request.RequestUri.AbsolutePath.EndsWith("/Get")) Interlocked.Increment(ref mutations); return OkMutationOrGet(request); });
|
|
var (session, owner) = await Owner("101", handler);
|
|
for (var i = 0; i < 51; i++) owner.Update("k" + i.ToString("D2"), "v");
|
|
var result = await owner.FlushAsync();
|
|
Check(result.Confirmed == 51 && mutations == 6, "51 keys in six batches");
|
|
await owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task WatermarkAndInFlightVersion()
|
|
{
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var mutation = 0;
|
|
var handler = new JsonHandler(async request =>
|
|
{
|
|
if (request.RequestUri.AbsolutePath.EndsWith("/Get")) return OkGet();
|
|
if (Interlocked.Increment(ref mutation) == 1) { entered.TrySetResult(true); await release.Task; }
|
|
return Ok();
|
|
});
|
|
var (session, owner) = await Owner("101", handler);
|
|
var v1 = owner.Update("same", "v1").Ticket;
|
|
var firstFlush = owner.FlushAsync(); await entered.Task;
|
|
var v2 = owner.Update("same", "v2").Ticket;
|
|
release.TrySetResult(true); var first = await firstFlush;
|
|
Check(first.Confirmed == 1 && (await owner.WaitForTicketAsync(v1)).Status == PlayerKVTicketStatus.Confirmed, "v1 confirmed");
|
|
using (var cts = new CancellationTokenSource(30))
|
|
Check((await owner.WaitForTicketAsync(v2, cts.Token)).Status == PlayerKVTicketStatus.Cancelled, "v2 outside watermark");
|
|
var secondFlush = await owner.FlushAsync();
|
|
Check(secondFlush.Confirmed == 1, "second summary excludes historical ticket");
|
|
Check((await owner.WaitForTicketAsync(v2)).Status == PlayerKVTicketStatus.Confirmed, "v2 next flush");
|
|
await owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task FlushWaitCancellation()
|
|
{
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var handler = new JsonHandler(async request =>
|
|
{ if (request.RequestUri.AbsolutePath.EndsWith("/Get")) return OkGet(); entered.TrySetResult(true); await release.Task; return Ok(); });
|
|
var (session, owner) = await Owner("101", handler); var ticket = owner.Update("a", "v").Ticket;
|
|
using var cts = new CancellationTokenSource(); var wait = owner.FlushAsync(cts.Token); await entered.Task; cts.Cancel();
|
|
Check((await wait).TransportError == PlayerKVError.Cancelled, "caller wait cancelled");
|
|
release.TrySetResult(true);
|
|
Check((await owner.WaitForTicketAsync(ticket)).Status == PlayerKVTicketStatus.Confirmed, "shared flush continued");
|
|
await owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task NetworkFailureUnknown()
|
|
{
|
|
var handler = new JsonHandler(request => request.RequestUri.AbsolutePath.EndsWith("/Get")
|
|
? Task.FromResult(OkGet()) : Task.FromException<HttpResponseMessage>(new HttpRequestException("synthetic")));
|
|
var (session, owner) = await Owner("101", handler); var ticket = owner.Update("a", "v").Ticket;
|
|
var flush = await owner.FlushAsync();
|
|
Check(flush.TransportError == PlayerKVError.Network && (await owner.WaitForTicketAsync(ticket)).Status == PlayerKVTicketStatus.Unknown, "network unknown");
|
|
await owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task CloseIdempotent()
|
|
{
|
|
var (session, owner) = await Owner("101", new JsonHandler(_ => OkMutationOrGet(_)));
|
|
var one = owner.CloseAsync(); var two = owner.CloseAsync();
|
|
Check(ReferenceEquals(one, two), "same close task"); await one; session.Dispose();
|
|
}
|
|
|
|
static async Task CloseWaitsForTransport()
|
|
{
|
|
var root = Root(); var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var session = await Login("101");
|
|
var opened = await PlayerKVSessionOwner.CreateAsync(session, root, "101", new Uri("http://unit.test/"), 1, _ => true,
|
|
default, _ => new JsonHandler(async request =>
|
|
{ if (request.RequestUri.AbsolutePath.EndsWith("/Get")) return OkGet(); entered.TrySetResult(true); await release.Task; return Ok(); }));
|
|
opened.Owner.Update("a", "v"); var flush = opened.Owner.FlushAsync(); await entered.Task;
|
|
var close = opened.Owner.CloseAsync();
|
|
Check(!close.IsCompleted, "close waits");
|
|
var busy = await PlayerKVSessionOwner.CreateAsync(session, root, "101", new Uri("http://unit.test/"), 2, _ => true,
|
|
default, _ => new JsonHandler(_ => OkGet()));
|
|
Check(busy.Error == PlayerKVSessionOpenError.Busy, "lock retained");
|
|
release.TrySetResult(true); await close; await flush; session.Dispose();
|
|
}
|
|
|
|
static async Task StaleEpoch()
|
|
{
|
|
var current = true; var session = await Login("101");
|
|
var opened = await PlayerKVSessionOwner.CreateAsync(session, Root(), "101", new Uri("http://unit.test/"), 4, _ => current,
|
|
default, _ => new JsonHandler(_ => OkGet()));
|
|
current = false;
|
|
Check(!opened.Owner.Update("a", "v").Accepted, "stale edit blocked");
|
|
await opened.Owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task ActiveWriteDiskFailure()
|
|
{
|
|
var armed = false; var mutations = 0;
|
|
var root = Root();
|
|
var opened = PlayerKVSyncState.Open(root, new Uri("http://unit.test/"), "101",
|
|
stage => { if (armed && stage == KVCacheWriteStage.BeforeWrite) throw new IOException("synthetic"); });
|
|
var http = new HttpClient(new JsonHandler(request =>
|
|
{ if (!request.RequestUri.AbsolutePath.EndsWith("/Get")) mutations++; return OkMutationOrGet(request); }));
|
|
var coordinator = new PlayerKVSyncCoordinator(opened.State, new PlayerKVClient(http, new Uri("http://unit.test/")),
|
|
http, "disk", 1, () => true, CancellationToken.None);
|
|
Check((await coordinator.InitializeAsync()).Success, "initialize");
|
|
var ticket = coordinator.Update("a", "v", true).Ticket; armed = true;
|
|
var watermark = coordinator.CaptureWatermark(out var targets);
|
|
var result = await coordinator.FlushThroughAsync(watermark, targets);
|
|
Check(mutations == 0 && result.Blocked == 1, "no send after active WAL failure");
|
|
Check((await coordinator.WaitForTicketAsync(ticket, default)).Status == PlayerKVTicketStatus.Blocked, "blocked ticket");
|
|
coordinator.Dispose(); opened.State.Dispose();
|
|
}
|
|
|
|
static async Task ConfirmDiskFailure()
|
|
{
|
|
var armed = false;
|
|
var root = Root();
|
|
var opened = PlayerKVSyncState.Open(root, new Uri("http://unit.test/"), "101",
|
|
stage => { if (armed && stage == KVCacheWriteStage.BeforeReplace) throw new IOException("synthetic"); });
|
|
var http = new HttpClient(new JsonHandler(request => OkMutationOrGet(request)));
|
|
var coordinator = new PlayerKVSyncCoordinator(opened.State, new PlayerKVClient(http, new Uri("http://unit.test/")),
|
|
http, "confirm", 1, () => true, CancellationToken.None);
|
|
Check((await coordinator.InitializeAsync()).Success, "initialize");
|
|
var ticket = coordinator.Update("a", "v", true).Ticket;
|
|
// BeginBatch also persists; arm only after its BeforeReplace by arming in the handler on dispatch.
|
|
coordinator.Dispose(); opened.State.Dispose();
|
|
|
|
opened = PlayerKVSyncState.Open(Root(), new Uri("http://unit.test/"), "101",
|
|
stage => { if (armed && stage == KVCacheWriteStage.BeforeReplace) throw new IOException("synthetic"); });
|
|
http = new HttpClient(new JsonHandler(request =>
|
|
{ if (!request.RequestUri.AbsolutePath.EndsWith("/Get")) armed = true; return OkMutationOrGet(request); }));
|
|
coordinator = new PlayerKVSyncCoordinator(opened.State, new PlayerKVClient(http, new Uri("http://unit.test/")),
|
|
http, "confirm2", 1, () => true, CancellationToken.None);
|
|
Check((await coordinator.InitializeAsync()).Success, "initialize2");
|
|
ticket = coordinator.Update("a", "v", true).Ticket;
|
|
var watermark = coordinator.CaptureWatermark(out var targets);
|
|
var result = await coordinator.FlushThroughAsync(watermark, targets);
|
|
Check(result.Unknown == 1 && opened.State.Status == PlayerKVSyncStatus.StorageError, "confirmation unknown");
|
|
Check((await coordinator.WaitForTicketAsync(ticket, default)).Status == PlayerKVTicketStatus.Unknown, "unknown ticket");
|
|
coordinator.Dispose(); opened.State.Dispose();
|
|
}
|
|
|
|
static async Task UnauthorizedWrite()
|
|
{
|
|
var handler = new JsonHandler(request => request.RequestUri.AbsolutePath.EndsWith("/Get")
|
|
? OkGet() : new HttpResponseMessage(HttpStatusCode.Unauthorized));
|
|
var (session, owner) = await Owner("101", handler); var ticket = owner.Update("a", "v").Ticket;
|
|
var result = await owner.FlushAsync();
|
|
Check(result.TransportError == PlayerKVError.Unauthorized && result.Unknown == 1, "owned unauthorized");
|
|
Check((await owner.WaitForTicketAsync(ticket)).Status == PlayerKVTicketStatus.Unknown, "unknown");
|
|
await owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task QueuedFlushClose()
|
|
{
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var handler = new JsonHandler(async request =>
|
|
{ if (request.RequestUri.AbsolutePath.EndsWith("/Get")) return OkGet(); entered.TrySetResult(true); await release.Task; return Ok(); });
|
|
var (session, owner) = await Owner("101", handler);
|
|
owner.Update("a", "v1"); var first = owner.FlushAsync(); await entered.Task;
|
|
owner.Update("b", "v2"); var queued = owner.FlushAsync();
|
|
var close = owner.CloseAsync(); release.TrySetResult(true);
|
|
await close; await first;
|
|
var queuedResult = await queued;
|
|
Check(queuedResult.TransportError == PlayerKVError.Cancelled, "queued flush structured cancellation");
|
|
session.Dispose();
|
|
}
|
|
|
|
static async Task QueuedRefreshClose()
|
|
{
|
|
var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var handler = new JsonHandler(async request =>
|
|
{
|
|
if (request.RequestUri.AbsolutePath.EndsWith("/Get")) return OkGet();
|
|
entered.TrySetResult(true); await release.Task; return Ok();
|
|
});
|
|
var (session, owner) = await Owner("101", handler);
|
|
owner.Update("a", "v"); var flush = owner.FlushAsync(); await entered.Task;
|
|
var queuedRefresh = owner.RefreshAsync();
|
|
var close = owner.CloseAsync(); release.TrySetResult(true);
|
|
await close; await flush;
|
|
var refreshResult = await queuedRefresh;
|
|
Check(refreshResult.Error == PlayerKVError.Cancelled, "queued refresh structured cancellation");
|
|
session.Dispose();
|
|
}
|
|
|
|
static async Task CreateReadErrors()
|
|
{
|
|
var session = await Login("101");
|
|
var protocol = await PlayerKVSessionOwner.CreateAsync(session, Root(), "101", new Uri("http://unit.test/"), 1, _ => true,
|
|
default, _ => new JsonHandler(_ => Json(HttpStatusCode.OK, "{}")));
|
|
Check(protocol.Error == PlayerKVSessionOpenError.ReadFailed && protocol.TransportError == PlayerKVError.Protocol, "protocol");
|
|
var network = await PlayerKVSessionOwner.CreateAsync(session, Root(), "101", new Uri("http://unit.test/"), 1, _ => true,
|
|
default, _ => new JsonHandler(_ => Task.FromException<HttpResponseMessage>(new HttpRequestException("synthetic"))));
|
|
Check(network.Error == PlayerKVSessionOpenError.ReadFailed && network.TransportError == PlayerKVError.Network, "network");
|
|
using var cts = new CancellationTokenSource(); cts.Cancel();
|
|
var cancelled = await PlayerKVSessionOwner.CreateAsync(session, Root(), "101", new Uri("http://unit.test/"), 1, _ => true,
|
|
cts.Token, _ => new JsonHandler(_ => OkGet()));
|
|
Check(cancelled.Error == PlayerKVSessionOpenError.Cancelled, "cancelled"); session.Dispose();
|
|
}
|
|
|
|
static async Task RecoveryRequired()
|
|
{
|
|
var root = Root(); var session = await Login("101");
|
|
var first = await PlayerKVSessionOwner.CreateAsync(session, root, "101", new Uri("http://unit.test/"), 1, _ => true,
|
|
default, _ => new JsonHandler(_ => OkGet()));
|
|
Check(first.Success, "first"); await first.Owner.CloseAsync();
|
|
var statePath = Directory.GetFiles(root, "state.json", SearchOption.AllDirectories).Single();
|
|
File.WriteAllText(statePath, "broken");
|
|
var second = await PlayerKVSessionOwner.CreateAsync(session, root, "101", new Uri("http://unit.test/"), 2, _ => true,
|
|
default, _ => new JsonHandler(_ => OkGet()));
|
|
Check(second.Error == PlayerKVSessionOpenError.RecoveryRequired, "recovery required"); session.Dispose();
|
|
}
|
|
|
|
static async Task LoginSessionInvalidation()
|
|
{
|
|
var (session, owner) = await Owner("101", new JsonHandler(_ => OkMutationOrGet(_)));
|
|
session.Dispose();
|
|
Check(!owner.Update("a", "v").Accepted, "edit blocked after login session dispose");
|
|
Check((await owner.RefreshAsync()).Error == PlayerKVError.Cancelled, "get blocked after invalidation");
|
|
await owner.CloseAsync();
|
|
}
|
|
|
|
static async Task MutationFailureVariants()
|
|
{
|
|
foreach (var code in new[] { HttpStatusCode.BadRequest, HttpStatusCode.Forbidden })
|
|
{
|
|
var handler = new JsonHandler(request => request.RequestUri.AbsolutePath.EndsWith("/Get") ? OkGet() : new HttpResponseMessage(code));
|
|
var (session, owner) = await Owner("101", handler); var ticket = owner.Update("a", "v").Ticket;
|
|
var result = await owner.FlushAsync();
|
|
Check(result.Unknown == 1, code + " unknown");
|
|
Check((await owner.WaitForTicketAsync(ticket)).Status == PlayerKVTicketStatus.Unknown, code + " ticket");
|
|
await owner.CloseAsync(); session.Dispose();
|
|
}
|
|
var timeoutSession = await Login("101");
|
|
var timeoutOwner = await PlayerKVSessionOwner.CreateAsync(timeoutSession, Root(), "101", new Uri("http://unit.test/"), 1, _ => true,
|
|
default, _ => new CancellationHandler(), TimeSpan.FromMilliseconds(30));
|
|
Check(timeoutOwner.Success, "timeout owner"); var timeoutTicket = timeoutOwner.Owner.Update("a", "v").Ticket;
|
|
var timeoutResult = await timeoutOwner.Owner.FlushAsync();
|
|
Check(timeoutResult.TransportError == PlayerKVError.Timeout && timeoutResult.Unknown == 1, "timeout unknown");
|
|
Check((await timeoutOwner.Owner.WaitForTicketAsync(timeoutTicket)).Status == PlayerKVTicketStatus.Unknown, "timeout ticket");
|
|
await timeoutOwner.Owner.CloseAsync(); timeoutSession.Dispose();
|
|
}
|
|
|
|
static async Task CancelCallbackReentry()
|
|
{
|
|
var session = await Login("101"); var handler = new ReentrantCancelHandler();
|
|
var opened = await PlayerKVSessionOwner.CreateAsync(session, Root(), "101", new Uri("http://unit.test/"), 1, _ => true,
|
|
default, _ => handler);
|
|
Check(opened.Success, "open");
|
|
handler.OnCancel = () => Check(!opened.Owner.Update("reentrant", "v").Accepted, "reentrant update blocked");
|
|
opened.Owner.Update("a", "v"); var flush = opened.Owner.FlushAsync(); await handler.Entered.Task;
|
|
await opened.Owner.CloseAsync(); await flush; session.Dispose();
|
|
}
|
|
|
|
static async Task RecoveredPendingCount()
|
|
{
|
|
var root = Root(); var session = await Login("101");
|
|
var first = await PlayerKVSessionOwner.CreateAsync(session, root, "101", new Uri("http://unit.test/"), 1, _ => true,
|
|
default, _ => new JsonHandler(_ => OkMutationOrGet(_)));
|
|
first.Owner.Update("persisted", "v"); await first.Owner.CloseAsync();
|
|
var second = await PlayerKVSessionOwner.CreateAsync(session, root, "101", new Uri("http://unit.test/"), 2, _ => true,
|
|
default, _ => new JsonHandler(_ => OkMutationOrGet(_)));
|
|
Check(second.Success, "reopen");
|
|
Check((await second.Owner.FlushAsync()).Confirmed == 1, "recovered pending count");
|
|
await second.Owner.CloseAsync(); session.Dispose();
|
|
}
|
|
|
|
static async Task<S101AccountSession> Login(string id)
|
|
{
|
|
var session = new S101AccountSession(new Uri("http://unit.test/"), jar => new JsonHandler(_ =>
|
|
{
|
|
jar.Add(new Uri("http://unit.test/"), new Cookie("sid", id));
|
|
return Json(HttpStatusCode.OK, "{\"state\":0,\"internalId\":1,\"newCreate\":false,\"createTime\":\"2026-01-01T00:00:00Z\",\"svrTime\":\"2026-01-01T00:00:00Z\"}");
|
|
}));
|
|
Check((await session.LoginAsync(id)).Success, "login"); return session;
|
|
}
|
|
|
|
static async Task<(S101AccountSession, PlayerKVSessionOwner)> Owner(string id, HttpMessageHandler handler)
|
|
{
|
|
var session = await Login(id);
|
|
var opened = await PlayerKVSessionOwner.CreateAsync(session, Root(), id, new Uri("http://unit.test/"), 1,
|
|
e => e == 1, default, _ => handler);
|
|
Check(opened.Success, "open " + opened.Error); return (session, opened.Owner);
|
|
}
|
|
|
|
static string Root() { var path = Path.Combine(Path.GetTempPath(), "kv3b-" + Guid.NewGuid().ToString("N")); Directory.CreateDirectory(path); return path; }
|
|
static HttpResponseMessage OkGet() => Json(HttpStatusCode.OK, "{\"data\":{}}");
|
|
static Task<HttpResponseMessage> OkMutationOrGet(HttpRequestMessage r) => Task.FromResult(r.RequestUri.AbsolutePath.EndsWith("/Get") ? OkGet() : Ok());
|
|
static HttpResponseMessage Ok() => new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent("{}", Encoding.UTF8, "application/json") };
|
|
static HttpResponseMessage Json(HttpStatusCode code, string json) => new HttpResponseMessage(code) { Content = new StringContent(json, Encoding.UTF8, "application/json") };
|
|
static void Check(bool value, string message) { if (!value) throw new Exception(message); }
|
|
|
|
sealed class JsonHandler : HttpMessageHandler
|
|
{
|
|
readonly Func<HttpRequestMessage, Task<HttpResponseMessage>> action;
|
|
int inFlight, max;
|
|
internal int MaxInFlight => max;
|
|
internal JsonHandler(Func<HttpRequestMessage, HttpResponseMessage> action) : this(r => Task.FromResult(action(r))) { }
|
|
internal JsonHandler(Func<HttpRequestMessage, Task<HttpResponseMessage>> action) { this.action = action; }
|
|
protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken token)
|
|
{
|
|
var count = Interlocked.Increment(ref inFlight); int old;
|
|
while (count > (old = max)) Interlocked.CompareExchange(ref max, count, old);
|
|
try { return await action(request); }
|
|
finally { Interlocked.Decrement(ref inFlight); }
|
|
}
|
|
}
|
|
|
|
sealed class CancellationHandler : HttpMessageHandler
|
|
{
|
|
protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken token)
|
|
{
|
|
if (request.RequestUri.AbsolutePath.EndsWith("/Get")) return OkGet();
|
|
await Task.Delay(Timeout.InfiniteTimeSpan, token); return Ok();
|
|
}
|
|
}
|
|
|
|
sealed class ReentrantCancelHandler : HttpMessageHandler
|
|
{
|
|
internal readonly TaskCompletionSource<bool> Entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
internal Action OnCancel;
|
|
protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken token)
|
|
{
|
|
if (request.RequestUri.AbsolutePath.EndsWith("/Get")) return OkGet();
|
|
Entered.TrySetResult(true);
|
|
var cancelled = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
using (token.Register(() => { OnCancel?.Invoke(); cancelled.TrySetResult(true); })) await cancelled.Task;
|
|
token.ThrowIfCancellationRequested(); return Ok();
|
|
}
|
|
}
|
|
}
|