740 lines
30 KiB
C#
740 lines
30 KiB
C#
using System;
|
|
using System.Collections.Generic;
|
|
using System.Collections.ObjectModel;
|
|
using System.IO;
|
|
using System.Linq;
|
|
|
|
namespace Core.KV
|
|
{
|
|
/// <summary>
|
|
/// Synchronous, fixed-identity cache state. No HTTP, gameplay initialization or timers.
|
|
/// All transitions and disk writes share a gate. The phase-3B owner must drain its
|
|
/// admitted network operations before disposing this state and its filesystem lease.
|
|
/// </summary>
|
|
public sealed class KvSyncState : IDisposable
|
|
{
|
|
private readonly object gate = new object();
|
|
private readonly KvCacheStore store;
|
|
private KvSnapshot snapshot;
|
|
private KVSyncStatus status;
|
|
private readonly HashSet<long> recoveredImmediate;
|
|
private bool recoverImmediateView = true;
|
|
private bool recoveredOrdinaryUnknown;
|
|
private KvSnapshot comparisonLocal;
|
|
private Dictionary<string, string> comparisonRemote;
|
|
private Dictionary<string, DateTime> comparisonTimes;
|
|
private Dictionary<string, string> comparisonView;
|
|
private HashSet<string> comparisonImmediateKeys;
|
|
internal bool HasPendingComparison
|
|
{
|
|
get
|
|
{
|
|
lock (gate)
|
|
{
|
|
return comparisonLocal != null;
|
|
}
|
|
}
|
|
}
|
|
internal bool AuthenticationInvalid
|
|
{
|
|
get
|
|
{
|
|
lock (gate)
|
|
{
|
|
return snapshot.AuthenticationInvalid;
|
|
}
|
|
}
|
|
}
|
|
public KVSyncStatus Status
|
|
{
|
|
get
|
|
{
|
|
lock (gate)
|
|
return status;
|
|
}
|
|
}
|
|
|
|
private KvSyncState(KvCacheStore store, KvSnapshot snapshot)
|
|
{
|
|
this.store = store;
|
|
this.snapshot = snapshot;
|
|
recoveredImmediate = new HashSet<long>(snapshot.RemoteOnlyWrites.Keys);
|
|
// Authentication belongs to the old owner. A new owner must obtain a fresh
|
|
// successful Get before Ready; this never clears a persisted unknown batch.
|
|
snapshot.AuthenticationInvalid = false;
|
|
// A recovered active batch might have reached the server. Never acknowledge it by guessing.
|
|
if (snapshot.Active != null)
|
|
{
|
|
if (snapshot.Active.IsOrdinary)
|
|
{
|
|
snapshot.OrdinaryUnknownBatches.Add(snapshot.Active);
|
|
}
|
|
else
|
|
{
|
|
snapshot.Unknown = snapshot.Active;
|
|
}
|
|
snapshot.Active = null;
|
|
}
|
|
recoveredOrdinaryUnknown = snapshot.OrdinaryUnknownBatches.Count != 0;
|
|
status = snapshot.Unknown != null || recoveredOrdinaryUnknown
|
|
? KVSyncStatus.UnknownOutcome : KVSyncStatus.NotLoaded;
|
|
}
|
|
|
|
public static KvOpenResult Open(string root, Uri api, string playerId)
|
|
{
|
|
KvIdentity identity;
|
|
try
|
|
{
|
|
identity = new KvIdentity(root, api, playerId);
|
|
}
|
|
catch (Exception ex) when (ex is ArgumentException || ex is NotSupportedException || ex is PathTooLongException)
|
|
{
|
|
return new KvOpenResult(KVOpenError.Validation);
|
|
}
|
|
var error = KvCacheStore.Open(identity, out var store, out var snapshot);
|
|
return error == KVOpenError.None ? new KvOpenResult(error, new KvSyncState(store, snapshot)) : new KvOpenResult(error);
|
|
}
|
|
|
|
public KvDiagnosticSnapshot Inspect()
|
|
{
|
|
lock (gate)
|
|
return new KvDiagnosticSnapshot(snapshot);
|
|
}
|
|
|
|
public IReadOnlyDictionary<string, string> ReadLocal()
|
|
{
|
|
lock (gate)
|
|
{
|
|
if (status != KVSyncStatus.Ready)
|
|
{
|
|
throw new InvalidOperationException("KV data is not ready.");
|
|
}
|
|
return new ReadOnlyDictionary<string, string>(ReadView());
|
|
}
|
|
}
|
|
|
|
// The caller must supply a successful complete Get. Failed reads must use RecordReadFailure.
|
|
|
|
internal KvStateResult Reconcile(IReadOnlyDictionary<string, string> remote,
|
|
IReadOnlyDictionary<string, DateTime> updatedAtUtc)
|
|
{
|
|
lock (gate)
|
|
{
|
|
if (Terminal || snapshot.Active != null || snapshot.AuthenticationInvalid || comparisonLocal != null)
|
|
{
|
|
return Blocked();
|
|
}
|
|
if (remote == null || remote.Any(pair => !KvSnapshot.ValidKey(pair.Key) || pair.Value == null))
|
|
{
|
|
return Invalid();
|
|
}
|
|
if (!recoverImmediateView || updatedAtUtc == null ||
|
|
updatedAtUtc.Count != remote.Count || remote.Keys.Any(key => !updatedAtUtc.ContainsKey(key)) ||
|
|
updatedAtUtc.Values.Any(time => time.Kind != DateTimeKind.Utc))
|
|
{
|
|
return Invalid();
|
|
}
|
|
var original = snapshot;
|
|
var copy = snapshot.Clone();
|
|
copy.Conflicts.Clear();
|
|
if (copy.Unknown != null)
|
|
{
|
|
foreach (var item in copy.Unsent())
|
|
{
|
|
copy.Conflicts[item.Key] = KvValue.At(remote, item.Key);
|
|
}
|
|
return Commit(copy, KVSyncStatus.UnknownOutcome);
|
|
}
|
|
var recoveringKeys = new HashSet<string>(copy.ImmediatePreservedKeys, StringComparer.Ordinal);
|
|
foreach (var sequence in recoveredImmediate)
|
|
{
|
|
var source = copy.RemoteOnlyWrites[sequence];
|
|
Rebase(copy, source.Key, KvValue.At(remote, source.Key));
|
|
}
|
|
foreach (var source in Sources(copy))
|
|
{
|
|
if (copy.OrdinaryBuffers.Contains(source))
|
|
{
|
|
continue;
|
|
}
|
|
foreach (var item in source.Values.ToArray())
|
|
{
|
|
var observed = KvValue.At(remote, item.Key);
|
|
if (observed.Equals(item.Target))
|
|
{
|
|
source.Remove(item.Key);
|
|
}
|
|
else if (!observed.Equals(item.BaseValue))
|
|
{
|
|
copy.Conflicts[item.Key] = observed;
|
|
}
|
|
}
|
|
}
|
|
foreach (var item in copy.PreparedOrdinaryBatches.SelectMany(batch => batch.Items))
|
|
{
|
|
var observed = KvValue.At(remote, item.Key);
|
|
if (!observed.Equals(item.Target) && !observed.Equals(item.BaseValue))
|
|
{
|
|
copy.Conflicts[item.Key] = observed;
|
|
}
|
|
}
|
|
var visible = new Dictionary<string, string>(remote, StringComparer.Ordinal);
|
|
foreach (var pair in copy.LocalOrigins.ToArray())
|
|
{
|
|
if (pair.Value.LocalOnly)
|
|
{
|
|
if (copy.LegacyLocalOnlyUnderImmediate.TryGetValue(pair.Key, out var lower))
|
|
{
|
|
visible[pair.Key] = lower;
|
|
}
|
|
else
|
|
{
|
|
Apply(visible, pair.Key, KvValue.At(copy.LocalData, pair.Key));
|
|
}
|
|
}
|
|
else
|
|
{
|
|
var source = copy.Unsent().Concat(copy.PreparedOrdinaryBatches.SelectMany(batch => batch.Items))
|
|
.Concat(copy.OrdinaryUnknownBatches.SelectMany(batch => batch.Items))
|
|
.FirstOrDefault(item => item.Sequence == pair.Value.Sequence);
|
|
if (source != null)
|
|
{
|
|
Apply(visible, pair.Key, recoveringKeys.Contains(pair.Key)
|
|
? source.Target : KvValue.At(copy.LocalData, pair.Key));
|
|
}
|
|
else
|
|
{
|
|
copy.LocalOrigins.Remove(pair.Key);
|
|
}
|
|
}
|
|
}
|
|
copy.Baseline = remote.ToDictionary(pair => pair.Key, pair => pair.Value, StringComparer.Ordinal);
|
|
copy.HasBaseline = true;
|
|
var reconciled = Commit(copy, copy.Conflicts.Count == 0 ? KVSyncStatus.Ready : KVSyncStatus.Conflict);
|
|
if (reconciled.Success)
|
|
{
|
|
recoverImmediateView = false;
|
|
comparisonLocal = original;
|
|
comparisonRemote = remote.ToDictionary(pair => pair.Key, pair => pair.Value, StringComparer.Ordinal);
|
|
comparisonTimes = updatedAtUtc.ToDictionary(pair => pair.Key, pair => pair.Value, StringComparer.Ordinal);
|
|
comparisonView = visible;
|
|
comparisonImmediateKeys = new HashSet<string>(StringComparer.Ordinal);
|
|
}
|
|
return reconciled;
|
|
}
|
|
}
|
|
|
|
private Dictionary<string, string> ReadView()
|
|
{
|
|
if (comparisonLocal == null)
|
|
{
|
|
return LocalView(snapshot);
|
|
}
|
|
var view = new Dictionary<string, string>(comparisonView, StringComparer.Ordinal);
|
|
foreach (var pair in snapshot.LocalOrigins)
|
|
{
|
|
if (pair.Value.Sequence >= comparisonLocal.NextSequence)
|
|
{
|
|
Apply(view, pair.Key, KvValue.At(snapshot.LocalData, pair.Key));
|
|
}
|
|
}
|
|
return view;
|
|
}
|
|
|
|
internal KvStateResult CompareDatas(long nowTicks)
|
|
{
|
|
lock (gate)
|
|
{
|
|
if (status != KVSyncStatus.Ready || snapshot.AuthenticationInvalid || snapshot.Active != null ||
|
|
snapshot.Unknown != null || comparisonLocal == null)
|
|
{
|
|
return Blocked();
|
|
}
|
|
if ((comparisonLocal.OrdinaryUnknownBatches.Count != 0 && !recoveredOrdinaryUnknown) ||
|
|
snapshot.OrdinaryUnknownBatches.Any(batch => !comparisonLocal.OrdinaryUnknownBatches.Any(initial => initial.Id == batch.Id)))
|
|
{
|
|
return Blocked();
|
|
}
|
|
if (nowTicks < 0 || nowTicks > DateTime.MaxValue.Ticks)
|
|
{
|
|
return Invalid();
|
|
}
|
|
var copy = snapshot.Clone();
|
|
// A new login uses the original time comparison, not the old request's outcome.
|
|
// Retire only its initial ordinary evidence in the same commit as that comparison.
|
|
foreach (var batch in comparisonLocal.OrdinaryUnknownBatches)
|
|
{
|
|
copy.OrdinaryUnknownBatches.RemoveAll(current => current.Id == batch.Id);
|
|
foreach (var item in batch.Items)
|
|
{
|
|
if (copy.LocalOrigins.TryGetValue(item.Key, out var origin) && origin.Sequence == item.Sequence)
|
|
{
|
|
copy.LocalOrigins.Remove(item.Key);
|
|
}
|
|
}
|
|
}
|
|
var protectedKeys = new HashSet<string>(StringComparer.Ordinal);
|
|
foreach (var source in new[] { comparisonLocal, snapshot })
|
|
{
|
|
protectedKeys.UnionWith(source.PreparedOrdinaryBatches.SelectMany(batch => batch.Items).Select(item => item.Key));
|
|
protectedKeys.UnionWith(source.LegacyPending.Keys);
|
|
protectedKeys.UnionWith(source.ExplicitDeletes.Keys);
|
|
protectedKeys.UnionWith(source.ImmediatePreservedKeys);
|
|
protectedKeys.UnionWith(source.RemoteOnlyWrites.Values.Select(item => item.Key));
|
|
}
|
|
protectedKeys.UnionWith(snapshot.LocalOrigins.Where(pair => pair.Value.Sequence >= comparisonLocal.NextSequence).Select(pair => pair.Key));
|
|
var visible = ReadView();
|
|
foreach (var key in protectedKeys)
|
|
{
|
|
Apply(copy.LocalData, key, KvValue.At(visible, key));
|
|
}
|
|
foreach (var key in comparisonLocal.LocalData.Keys.Concat(comparisonRemote.Keys).Distinct(StringComparer.Ordinal))
|
|
{
|
|
if (protectedKeys.Contains(key))
|
|
{
|
|
continue;
|
|
}
|
|
var localExists = comparisonLocal.HasBaseline && comparisonLocal.LocalData.TryGetValue(key, out _);
|
|
comparisonLocal.LocalData.TryGetValue(key, out var localValue);
|
|
var remoteExists = comparisonRemote.TryGetValue(key, out var remoteValue);
|
|
var localTime = comparisonLocal.LocalUpdateTicks.TryGetValue(key, out var ticks) ? ticks : 0;
|
|
var remoteTime = remoteExists ? comparisonTimes[key].Ticks : 0;
|
|
if (!localExists || (remoteExists && remoteTime - localTime > TimeSpan.TicksPerSecond * 60))
|
|
{
|
|
copy.LocalData[key] = remoteValue;
|
|
copy.LocalUpdateTicks[key] = remoteTime;
|
|
foreach (var bucket in copy.OrdinaryBuffers)
|
|
{
|
|
if (bucket.TryGetValue(key, out var current) && comparisonLocal.OrdinaryBuffers.Any(old =>
|
|
old.TryGetValue(key, out var item) && item.Sequence == current.Sequence))
|
|
{
|
|
bucket.Remove(key);
|
|
if (copy.LocalOrigins.TryGetValue(key, out var origin) && origin.Sequence == current.Sequence)
|
|
{
|
|
copy.LocalOrigins.Remove(key);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
else if (!remoteExists || localTime - remoteTime > TimeSpan.TicksPerSecond * 60)
|
|
{
|
|
if (copy.NextSequence == long.MaxValue)
|
|
{
|
|
return Invalid();
|
|
}
|
|
ApplyUpdate(copy, key, localValue, true, nowTicks);
|
|
}
|
|
else
|
|
{
|
|
copy.LocalData[key] = localValue;
|
|
copy.LocalUpdateTicks[key] = localTime;
|
|
}
|
|
}
|
|
foreach (var sequence in comparisonLocal.RemoteOnlyWrites.Keys)
|
|
{
|
|
copy.RemoteOnlyWrites.Remove(sequence);
|
|
}
|
|
foreach (var key in comparisonLocal.LegacyLocalOnlyUnderImmediate.Keys)
|
|
{
|
|
copy.LegacyLocalOnlyUnderImmediate.Remove(key);
|
|
}
|
|
foreach (var key in comparisonLocal.ImmediatePreservedKeys)
|
|
{
|
|
if (!comparisonImmediateKeys.Contains(key))
|
|
{
|
|
copy.ImmediatePreservedKeys.Remove(key);
|
|
}
|
|
}
|
|
var result = Commit(copy, KVSyncStatus.Ready);
|
|
if (result.Success)
|
|
{
|
|
recoveredOrdinaryUnknown = false;
|
|
ClearComparison();
|
|
recoveredImmediate.Clear();
|
|
}
|
|
return result;
|
|
}
|
|
}
|
|
|
|
private void ClearComparison()
|
|
{
|
|
comparisonLocal = null;
|
|
comparisonRemote = null;
|
|
comparisonTimes = null;
|
|
comparisonView = null;
|
|
comparisonImmediateKeys = null;
|
|
}
|
|
|
|
public KvStateResult RecordReadFailure(bool unauthorized = false)
|
|
{
|
|
lock (gate)
|
|
{
|
|
if (Terminal || snapshot.Active != null)
|
|
{
|
|
return Blocked();
|
|
}
|
|
if (unauthorized)
|
|
{
|
|
var copy = snapshot.Clone();
|
|
copy.AuthenticationInvalid = true;
|
|
return Commit(copy, copy.Unknown == null ? KVSyncStatus.Unauthorized : KVSyncStatus.UnknownOutcome);
|
|
}
|
|
status = snapshot.Unknown != null ? KVSyncStatus.UnknownOutcome : KVSyncStatus.ReadFailed;
|
|
return new KvStateResult(KVChangeError.None);
|
|
}
|
|
}
|
|
|
|
internal KvStateResult Update(string key, string value, bool remoteStore, long updateTicks)
|
|
{
|
|
lock (gate)
|
|
{
|
|
if (status != KVSyncStatus.Ready || snapshot.AuthenticationInvalid)
|
|
{
|
|
return Blocked();
|
|
}
|
|
if (!KvSnapshot.ValidKey(key) || value == null || snapshot.NextSequence == long.MaxValue ||
|
|
(updateTicks < 0 || updateTicks > DateTime.MaxValue.Ticks))
|
|
{
|
|
return Invalid();
|
|
}
|
|
var copy = snapshot.Clone();
|
|
var sequence = ApplyUpdate(copy, key, value, remoteStore, updateTicks);
|
|
return Commit(copy, KVSyncStatus.Ready, sequence);
|
|
}
|
|
}
|
|
|
|
private static long ApplyUpdate(KvSnapshot copy, string key, string value, bool remoteStore, long updateTicks)
|
|
{
|
|
var sequence = copy.NextSequence++;
|
|
copy.LocalData[key] = value;
|
|
copy.ImmediatePreservedKeys.Remove(key);
|
|
copy.LegacyLocalOnlyUnderImmediate.Remove(key);
|
|
copy.LocalUpdateTicks[key] = updateTicks;
|
|
var acceptedRemote = false;
|
|
if (remoteStore)
|
|
{
|
|
foreach (var bucket in copy.OrdinaryBuffers)
|
|
{
|
|
if (!bucket.ContainsKey(key) && bucket.Count >= 10)
|
|
{
|
|
continue;
|
|
}
|
|
var baseline = bucket.TryGetValue(key, out var previous)
|
|
? previous.BaseValue : KvValue.At(copy.Baseline, key);
|
|
bucket[key] = new KvChange(key, sequence, baseline, KvValue.Present(value));
|
|
copy.LegacyPending.Remove(key);
|
|
copy.ExplicitDeletes.Remove(key);
|
|
acceptedRemote = true;
|
|
break;
|
|
}
|
|
}
|
|
copy.LocalOrigins[key] = (!acceptedRemote, sequence);
|
|
return sequence;
|
|
}
|
|
|
|
internal KvStateResult CaptureOrdinaryBatch(out KvBatch batch)
|
|
{
|
|
lock (gate)
|
|
{
|
|
batch = null;
|
|
if (status != KVSyncStatus.Ready || snapshot.AuthenticationInvalid)
|
|
{
|
|
return Blocked();
|
|
}
|
|
if (snapshot.OrdinaryBuffers[0].Count == 0)
|
|
{
|
|
return new KvStateResult(KVChangeError.None);
|
|
}
|
|
var copy = snapshot.Clone();
|
|
var first = copy.OrdinaryBuffers[0];
|
|
var captured = new KvBatch(Guid.NewGuid().ToString("N"), first.Values.OrderBy(item => item.Sequence), true);
|
|
copy.PreparedOrdinaryBatches.Add(captured);
|
|
copy.OrdinaryBuffers.RemoveAt(0);
|
|
copy.OrdinaryBuffers.Add(new Dictionary<string, KvChange>(StringComparer.Ordinal));
|
|
var result = Commit(copy, KVSyncStatus.Ready);
|
|
if (result.Success)
|
|
{
|
|
batch = captured;
|
|
}
|
|
return result;
|
|
}
|
|
}
|
|
|
|
internal KvStateResult BeginOrdinaryBatch(out KvBatch batch)
|
|
{
|
|
lock (gate)
|
|
{
|
|
batch = null;
|
|
if (status != KVSyncStatus.Ready || snapshot.Active != null || snapshot.Unknown != null || snapshot.AuthenticationInvalid)
|
|
{
|
|
return Blocked();
|
|
}
|
|
if (snapshot.PreparedOrdinaryBatches.Count == 0)
|
|
{
|
|
return new KvStateResult(KVChangeError.None);
|
|
}
|
|
var copy = snapshot.Clone();
|
|
copy.Active = copy.PreparedOrdinaryBatches[0];
|
|
copy.PreparedOrdinaryBatches.RemoveAt(0);
|
|
var result = Commit(copy, KVSyncStatus.Ready);
|
|
if (result.Success)
|
|
{
|
|
batch = snapshot.Active;
|
|
}
|
|
return result;
|
|
}
|
|
}
|
|
|
|
// Explicit requests retain their original source until the outcome is known. Ordinary buckets are never selected here.
|
|
|
|
internal KvStateResult BeginBatch(long throughSequence, out KvBatch batch)
|
|
{
|
|
lock (gate)
|
|
{
|
|
batch = null;
|
|
if (status != KVSyncStatus.Ready || snapshot.Active != null || snapshot.Unknown != null || snapshot.AuthenticationInvalid)
|
|
{
|
|
return Blocked();
|
|
}
|
|
if (throughSequence < 0)
|
|
{
|
|
return Invalid();
|
|
}
|
|
var source = snapshot.LegacyPending.Values;
|
|
var ordered = source.Where(item => item.Sequence <= throughSequence).OrderBy(item => item.Sequence).ToArray();
|
|
if (ordered.Length == 0)
|
|
{
|
|
return new KvStateResult(KVChangeError.None);
|
|
}
|
|
var copy = snapshot.Clone();
|
|
copy.Active = new KvBatch(Guid.NewGuid().ToString("N"), ordered.TakeWhile(item => item.Target.Exists == ordered[0].Target.Exists).Take(10));
|
|
var result = Commit(copy, KVSyncStatus.Ready);
|
|
if (result.Success)
|
|
{
|
|
batch = snapshot.Active;
|
|
}
|
|
return result;
|
|
}
|
|
}
|
|
|
|
public KvStateResult ConfirmBatch(string batchId)
|
|
{
|
|
lock (gate)
|
|
{
|
|
if (status != KVSyncStatus.Ready || snapshot.Active == null || snapshot.Active.Id != batchId)
|
|
{
|
|
return Blocked();
|
|
}
|
|
var copy = snapshot.Clone();
|
|
foreach (var item in copy.Active.Items)
|
|
{
|
|
Apply(copy.Baseline, item.Key, item.Target);
|
|
if (!copy.Active.IsOrdinary)
|
|
{
|
|
foreach (var source in new[] { copy.LegacyPending, copy.ExplicitDeletes })
|
|
{
|
|
if (source.TryGetValue(item.Key, out var current) && current.Sequence == item.Sequence)
|
|
{
|
|
source.Remove(item.Key);
|
|
}
|
|
}
|
|
}
|
|
Rebase(copy, item.Key, item.Target);
|
|
}
|
|
copy.Active = null;
|
|
return Commit(copy, KVSyncStatus.Ready);
|
|
}
|
|
}
|
|
|
|
// A definitely undispatched ordinary batch retires, as the original SDK call did; explicit source references stay.
|
|
public KvStateResult FailBatch(string batchId, bool mayHaveCommitted, bool unauthorized = false)
|
|
{
|
|
lock (gate)
|
|
{
|
|
if (Terminal || snapshot.Active == null || snapshot.Active.Id != batchId)
|
|
{
|
|
return Blocked();
|
|
}
|
|
var copy = snapshot.Clone();
|
|
var ordinary = copy.Active.IsOrdinary;
|
|
copy.AuthenticationInvalid |= unauthorized;
|
|
if (mayHaveCommitted)
|
|
{
|
|
if (ordinary)
|
|
{
|
|
copy.OrdinaryUnknownBatches.Add(copy.Active);
|
|
}
|
|
else
|
|
{
|
|
copy.Unknown = copy.Active;
|
|
}
|
|
}
|
|
copy.Active = null;
|
|
var next = copy.AuthenticationInvalid ? KVSyncStatus.Unauthorized : ordinary ? KVSyncStatus.Ready : KVSyncStatus.ReadFailed;
|
|
if (copy.Unknown != null)
|
|
{
|
|
next = KVSyncStatus.UnknownOutcome;
|
|
}
|
|
return Commit(copy, next);
|
|
}
|
|
}
|
|
|
|
private static IEnumerable<Dictionary<string, KvChange>> Sources(KvSnapshot value)
|
|
{
|
|
return value.OrdinaryBuffers.Concat(new[] { value.LegacyPending, value.ExplicitDeletes });
|
|
}
|
|
private static void RemoveSources(KvSnapshot value, string key)
|
|
{
|
|
foreach (var source in Sources(value))
|
|
{
|
|
source.Remove(key);
|
|
}
|
|
}
|
|
private static void Rebase(KvSnapshot value, string key, KvValue baseline)
|
|
{
|
|
foreach (var source in Sources(value))
|
|
{
|
|
if (source.TryGetValue(key, out var pending))
|
|
{
|
|
source[key] = pending.WithBase(baseline);
|
|
}
|
|
}
|
|
}
|
|
|
|
internal KvStateResult PrepareImmediate(IReadOnlyDictionary<string, string> data,
|
|
out IReadOnlyList<KvChange> operations)
|
|
{
|
|
lock (gate)
|
|
{
|
|
operations = null;
|
|
if (status != KVSyncStatus.Ready || snapshot.AuthenticationInvalid)
|
|
{
|
|
return Blocked();
|
|
}
|
|
if (data == null || data.Count == 0 || data.Count > long.MaxValue - snapshot.NextSequence ||
|
|
data.Any(pair => !KvSnapshot.ValidKey(pair.Key) || pair.Value == null))
|
|
{
|
|
return Invalid();
|
|
}
|
|
var copy = snapshot.Clone();
|
|
var prepared = new List<KvChange>();
|
|
foreach (var pair in data)
|
|
{
|
|
copy.ImmediatePreservedKeys.Add(pair.Key);
|
|
var operation = new KvChange(pair.Key, copy.NextSequence++,
|
|
KvValue.At(copy.Baseline, pair.Key), KvValue.Present(pair.Value));
|
|
copy.RemoteOnlyWrites.Add(operation.Sequence, operation);
|
|
RemoveSources(copy, pair.Key);
|
|
prepared.Add(operation);
|
|
}
|
|
var result = Commit(copy, KVSyncStatus.Ready, prepared[0].Sequence);
|
|
if (result.Success)
|
|
{
|
|
operations = prepared.AsReadOnly();
|
|
if (comparisonLocal != null)
|
|
{
|
|
comparisonImmediateKeys.UnionWith(data.Keys);
|
|
}
|
|
}
|
|
return result;
|
|
}
|
|
}
|
|
|
|
internal KvStateResult CompleteImmediate(IReadOnlyList<KvChange> operations)
|
|
{
|
|
lock (gate)
|
|
{
|
|
if (status != KVSyncStatus.Ready || snapshot.AuthenticationInvalid || snapshot.Active != null ||
|
|
snapshot.Unknown != null || operations.Any(operation => !snapshot.RemoteOnlyWrites.ContainsKey(operation.Sequence)))
|
|
{
|
|
return Blocked();
|
|
}
|
|
var copy = snapshot.Clone();
|
|
foreach (var operation in operations)
|
|
{
|
|
Apply(copy.Baseline, operation.Key, operation.Target);
|
|
Rebase(copy, operation.Key, operation.Target);
|
|
copy.RemoteOnlyWrites.Remove(operation.Sequence);
|
|
copy.ImmediatePreservedKeys.Add(operation.Key);
|
|
}
|
|
return Commit(copy, KVSyncStatus.Ready);
|
|
}
|
|
}
|
|
|
|
internal void CancelUndispatchedImmediate(IReadOnlyList<KvChange> operations)
|
|
{
|
|
lock (gate)
|
|
{
|
|
if (Terminal)
|
|
{
|
|
return;
|
|
}
|
|
var protectedBatches = snapshot.OrdinaryUnknownBatches.Concat(new[] { snapshot.Active, snapshot.Unknown }).Where(batch => batch != null);
|
|
var removable = operations.Where(operation => snapshot.RemoteOnlyWrites.ContainsKey(operation.Sequence) &&
|
|
!protectedBatches.Any(batch => batch.Items.Any(item => item.Key == operation.Key && item.Sequence < operation.Sequence))).ToArray();
|
|
if (removable.Length == 0)
|
|
{
|
|
return;
|
|
}
|
|
var copy = snapshot.Clone();
|
|
foreach (var operation in removable)
|
|
{
|
|
copy.RemoteOnlyWrites.Remove(operation.Sequence);
|
|
}
|
|
Commit(copy, status);
|
|
}
|
|
}
|
|
|
|
private static Dictionary<string, string> LocalView(KvSnapshot source)
|
|
{
|
|
return new Dictionary<string, string>(source.LocalData, StringComparer.Ordinal);
|
|
}
|
|
|
|
private bool Terminal => status == KVSyncStatus.Closed || status == KVSyncStatus.StorageError;
|
|
private KvStateResult Commit(KvSnapshot copy, KVSyncStatus nextStatus, long sequence = 0)
|
|
{
|
|
if (snapshot.Revision == long.MaxValue)
|
|
{
|
|
status = KVSyncStatus.StorageError;
|
|
return new KvStateResult(KVChangeError.StorageError, sequence);
|
|
}
|
|
copy.Revision = snapshot.Revision + 1;
|
|
if (!store.Save(copy, out var uncertain))
|
|
{
|
|
status = KVSyncStatus.StorageError;
|
|
return new KvStateResult(KVChangeError.StorageError, sequence, uncertain);
|
|
}
|
|
snapshot = copy;
|
|
status = nextStatus;
|
|
return new KvStateResult(KVChangeError.None, sequence);
|
|
}
|
|
private static void Apply(Dictionary<string, string> data, string key, KvValue value)
|
|
{
|
|
if (value.Exists)
|
|
{
|
|
data[key] = value.Value;
|
|
}
|
|
else
|
|
{
|
|
data.Remove(key);
|
|
}
|
|
}
|
|
private KvStateResult Blocked() => new KvStateResult(KVChangeError.Blocked);
|
|
private KvStateResult Invalid() => new KvStateResult(KVChangeError.Validation);
|
|
|
|
public void Dispose()
|
|
{
|
|
lock (gate)
|
|
{
|
|
if (status == KVSyncStatus.Closed)
|
|
{
|
|
return;
|
|
}
|
|
status = KVSyncStatus.Closed;
|
|
ClearComparison();
|
|
store.Dispose();
|
|
}
|
|
}
|
|
}
|
|
}
|