chore(m2): 保存功能验收及后续清理前检查点
This commit is contained in:
Liubing\LB
@@ -13,10 +13,38 @@ namespace game.PlayerDataStorage
|
||||
/// </summary>
|
||||
public sealed class PlayerKVSyncState : IDisposable
|
||||
{
|
||||
readonly object gate = new object();
|
||||
readonly PlayerKVCacheStore store;
|
||||
KVSnapshot snapshot;
|
||||
PlayerKVSyncStatus status;
|
||||
private readonly object gate = new object();
|
||||
private readonly PlayerKVCacheStore store;
|
||||
private KVSnapshot snapshot;
|
||||
private PlayerKVSyncStatus 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 PlayerKVSyncStatus Status { get { lock (gate) return status; } }
|
||||
public string ApiBaseUri { get; }
|
||||
public string PlayerId { get; }
|
||||
@@ -24,12 +52,26 @@ namespace game.PlayerDataStorage
|
||||
PlayerKVSyncState(KVIdentity identity, PlayerKVCacheStore store, KVSnapshot snapshot)
|
||||
{
|
||||
ApiBaseUri = identity.Api; PlayerId = identity.PlayerId; 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) { snapshot.Unknown = snapshot.Active; snapshot.Active = null; }
|
||||
status = snapshot.Unknown != null ? PlayerKVSyncStatus.UnknownOutcome : PlayerKVSyncStatus.NotLoaded;
|
||||
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
|
||||
? PlayerKVSyncStatus.UnknownOutcome : PlayerKVSyncStatus.NotLoaded;
|
||||
}
|
||||
|
||||
public static PlayerKVOpenResult Open(string root, Uri api, string playerId) => Open(root, api, playerId, null);
|
||||
@@ -50,39 +92,307 @@ namespace game.PlayerDataStorage
|
||||
lock (gate)
|
||||
{
|
||||
if (status != PlayerKVSyncStatus.Ready) throw new InvalidOperationException("KV data is not ready.");
|
||||
var result = new Dictionary<string, string>(snapshot.Baseline, StringComparer.Ordinal);
|
||||
foreach (var item in snapshot.Pending.Values) Apply(result, item.Key, item.Target);
|
||||
foreach (var pair in snapshot.LocalOnly) result[pair.Key] = pair.Value;
|
||||
return new ReadOnlyDictionary<string, string>(result);
|
||||
return new ReadOnlyDictionary<string, string>(ReadView());
|
||||
}
|
||||
}
|
||||
|
||||
// The caller must supply a successful complete Get. Failed reads must use RecordReadFailure.
|
||||
public PlayerKVStateResult Reconcile(IReadOnlyDictionary<string, string> remote)
|
||||
{
|
||||
return Reconcile(remote, null, false);
|
||||
}
|
||||
|
||||
internal PlayerKVStateResult Reconcile(IReadOnlyDictionary<string, string> remote,
|
||||
IReadOnlyDictionary<string, DateTime> updatedAtUtc, bool prepareOriginalComparison)
|
||||
{
|
||||
lock (gate)
|
||||
{
|
||||
if (Terminal || snapshot.Active != null || snapshot.AuthenticationInvalid) return Blocked();
|
||||
if (remote == null || remote.Any(p => !KVSnapshot.ValidKey(p.Key) || p.Value == null)) return Invalid();
|
||||
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 (prepareOriginalComparison && (!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)
|
||||
if (copy.Unknown != null || (recoveredOrdinaryUnknown && !prepareOriginalComparison))
|
||||
{
|
||||
foreach (var item in copy.Pending.Values) copy.Conflicts[item.Key] = PlayerKVValue.At(remote, item.Key);
|
||||
foreach (var item in copy.Unsent())
|
||||
{
|
||||
copy.Conflicts[item.Key] = PlayerKVValue.At(remote, item.Key);
|
||||
}
|
||||
return Commit(copy, PlayerKVSyncStatus.UnknownOutcome);
|
||||
}
|
||||
foreach (var item in copy.Pending.Values.ToArray())
|
||||
var recoveringKeys = recoverImmediateView
|
||||
? new HashSet<string>(copy.ImmediatePreservedKeys, StringComparer.Ordinal)
|
||||
: new HashSet<string>(StringComparer.Ordinal);
|
||||
if (recoverImmediateView)
|
||||
{
|
||||
foreach (var sequence in recoveredImmediate)
|
||||
{
|
||||
var source = copy.RemoteOnlyWrites[sequence];
|
||||
Rebase(copy, source.Key, PlayerKVValue.At(remote, source.Key));
|
||||
if (!prepareOriginalComparison)
|
||||
{
|
||||
copy.RemoteOnlyWrites.Remove(sequence);
|
||||
}
|
||||
}
|
||||
if (!prepareOriginalComparison)
|
||||
{
|
||||
copy.ImmediatePreservedKeys.Clear();
|
||||
}
|
||||
}
|
||||
foreach (var source in Sources(copy))
|
||||
{
|
||||
if (prepareOriginalComparison && copy.OrdinaryBuffers.Contains(source))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
foreach (var item in source.Values.ToArray())
|
||||
{
|
||||
var observed = PlayerKVValue.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 = PlayerKVValue.At(remote, item.Key);
|
||||
if (observed.Equals(item.Target)) copy.Pending.Remove(item.Key);
|
||||
else if (!observed.Equals(item.BaseValue)) copy.Conflicts[item.Key] = observed;
|
||||
if (!observed.Equals(item.Target) && !observed.Equals(item.BaseValue))
|
||||
{
|
||||
copy.Conflicts[item.Key] = observed;
|
||||
}
|
||||
}
|
||||
copy.Baseline = remote.ToDictionary(p => p.Key, p => p.Value, StringComparer.Ordinal);
|
||||
var visible = new Dictionary<string, string>(remote, StringComparer.Ordinal);
|
||||
foreach (var pair in copy.LocalOrigins.ToArray())
|
||||
{
|
||||
if (pair.Value.LocalOnly)
|
||||
{
|
||||
if (recoverImmediateView && copy.LegacyLocalOnlyUnderImmediate.TryGetValue(pair.Key, out var lower))
|
||||
{
|
||||
visible[pair.Key] = lower;
|
||||
}
|
||||
else
|
||||
{
|
||||
Apply(visible, pair.Key, PlayerKVValue.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 (!recoverImmediateView)
|
||||
{
|
||||
Apply(visible, pair.Key, PlayerKVValue.At(copy.LocalData, pair.Key));
|
||||
}
|
||||
else if (source != null)
|
||||
{
|
||||
Apply(visible, pair.Key, recoveringKeys.Contains(pair.Key)
|
||||
? source.Target : PlayerKVValue.At(copy.LocalData, pair.Key));
|
||||
}
|
||||
else
|
||||
{
|
||||
copy.LocalOrigins.Remove(pair.Key);
|
||||
}
|
||||
}
|
||||
}
|
||||
// An immediate write preserves its view until a new owner performs its initial Get.
|
||||
foreach (var key in copy.ImmediatePreservedKeys)
|
||||
{
|
||||
if (prepareOriginalComparison && recoveringKeys.Contains(key))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
Apply(visible, key, PlayerKVValue.At(copy.LocalData, key));
|
||||
}
|
||||
if (recoverImmediateView && !prepareOriginalComparison)
|
||||
{
|
||||
copy.LegacyLocalOnlyUnderImmediate.Clear();
|
||||
}
|
||||
if (!prepareOriginalComparison)
|
||||
{
|
||||
copy.LocalData = visible;
|
||||
}
|
||||
copy.Baseline = remote.ToDictionary(pair => pair.Key, pair => pair.Value, StringComparer.Ordinal);
|
||||
copy.HasBaseline = true;
|
||||
return Commit(copy, copy.Conflicts.Count == 0 ? PlayerKVSyncStatus.Ready : PlayerKVSyncStatus.Conflict);
|
||||
var reconciled = Commit(copy, copy.Conflicts.Count == 0 ? PlayerKVSyncStatus.Ready : PlayerKVSyncStatus.Conflict);
|
||||
if (reconciled.Success)
|
||||
{
|
||||
recoverImmediateView = false;
|
||||
if (prepareOriginalComparison)
|
||||
{
|
||||
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);
|
||||
}
|
||||
else
|
||||
{
|
||||
recoveredImmediate.Clear();
|
||||
}
|
||||
}
|
||||
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, PlayerKVValue.At(snapshot.LocalData, pair.Key));
|
||||
}
|
||||
}
|
||||
return view;
|
||||
}
|
||||
|
||||
internal PlayerKVStateResult CompareDatas(long nowTicks)
|
||||
{
|
||||
lock (gate)
|
||||
{
|
||||
if (status != PlayerKVSyncStatus.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, PlayerKVValue.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, PlayerKVSyncStatus.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 PlayerKVStateResult RecordReadFailure(bool unauthorized = false)
|
||||
{
|
||||
lock (gate)
|
||||
@@ -98,42 +408,177 @@ namespace game.PlayerDataStorage
|
||||
}
|
||||
}
|
||||
|
||||
public PlayerKVStateResult Update(string key, string value, bool remoteStore = true) =>
|
||||
Edit(key, value == null ? null : PlayerKVValue.Present(value), remoteStore);
|
||||
public PlayerKVStateResult Delete(string key) => Edit(key, PlayerKVValue.Missing, true);
|
||||
public PlayerKVStateResult Update(string key, string value, bool remoteStore = true)
|
||||
{
|
||||
return Update(key, value, remoteStore, null);
|
||||
}
|
||||
|
||||
PlayerKVStateResult Edit(string key, PlayerKVValue target, bool remote)
|
||||
internal PlayerKVStateResult Update(string key, string value, bool remoteStore, long? updateTicks)
|
||||
{
|
||||
lock (gate)
|
||||
{
|
||||
if (status != PlayerKVSyncStatus.Ready || snapshot.AuthenticationInvalid) return Blocked();
|
||||
if (!KVSnapshot.ValidKey(key) || target == null || snapshot.NextSequence == long.MaxValue) return Invalid();
|
||||
var copy = snapshot.Clone();
|
||||
var sequence = copy.NextSequence++;
|
||||
if (remote)
|
||||
if (status != PlayerKVSyncStatus.Ready || snapshot.AuthenticationInvalid)
|
||||
{
|
||||
var baseline = copy.Pending.TryGetValue(key, out var previous) ? previous.BaseValue : PlayerKVValue.At(copy.Baseline, key);
|
||||
copy.Pending[key] = new PlayerKVChange(key, sequence, baseline, target);
|
||||
copy.LocalOnly.Remove(key);
|
||||
return Blocked();
|
||||
}
|
||||
else copy.LocalOnly[key] = target.Value;
|
||||
if (!KVSnapshot.ValidKey(key) || value == null || snapshot.NextSequence == long.MaxValue ||
|
||||
(updateTicks.HasValue && (updateTicks.Value < 0 || updateTicks.Value > DateTime.MaxValue.Ticks)))
|
||||
{
|
||||
return Invalid();
|
||||
}
|
||||
var copy = snapshot.Clone();
|
||||
var sequence = ApplyUpdate(copy, key, value, remoteStore, updateTicks);
|
||||
return Commit(copy, PlayerKVSyncStatus.Ready, sequence);
|
||||
}
|
||||
}
|
||||
|
||||
public PlayerKVStateResult BeginBatch(long throughSequence, out PlayerKVBatch batch)
|
||||
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);
|
||||
if (updateTicks.HasValue)
|
||||
{
|
||||
copy.LocalUpdateTicks[key] = updateTicks.Value;
|
||||
}
|
||||
else
|
||||
{
|
||||
copy.LocalUpdateTicks.Remove(key);
|
||||
}
|
||||
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 : PlayerKVValue.At(copy.Baseline, key);
|
||||
bucket[key] = new PlayerKVChange(key, sequence, baseline, PlayerKVValue.Present(value));
|
||||
copy.LegacyPending.Remove(key);
|
||||
copy.ExplicitDeletes.Remove(key);
|
||||
acceptedRemote = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
copy.LocalOrigins[key] = (!acceptedRemote, sequence);
|
||||
return sequence;
|
||||
}
|
||||
|
||||
public PlayerKVStateResult Delete(string key)
|
||||
{
|
||||
lock (gate)
|
||||
{
|
||||
if (status != PlayerKVSyncStatus.Ready || snapshot.AuthenticationInvalid)
|
||||
{
|
||||
return Blocked();
|
||||
}
|
||||
if (!KVSnapshot.ValidKey(key) || snapshot.NextSequence == long.MaxValue)
|
||||
{
|
||||
return Invalid();
|
||||
}
|
||||
var copy = snapshot.Clone();
|
||||
var sequence = copy.NextSequence++;
|
||||
var prior = copy.ExplicitDeletes.TryGetValue(key, out var previous) ? previous.BaseValue : PlayerKVValue.At(copy.Baseline, key);
|
||||
RemoveSources(copy, key);
|
||||
copy.ExplicitDeletes[key] = new PlayerKVChange(key, sequence, prior, PlayerKVValue.Missing);
|
||||
copy.LocalData.Remove(key);
|
||||
copy.LocalUpdateTicks.Remove(key);
|
||||
copy.LocalOrigins[key] = (false, sequence);
|
||||
copy.ImmediatePreservedKeys.Remove(key);
|
||||
copy.LegacyLocalOnlyUnderImmediate.Remove(key);
|
||||
return Commit(copy, PlayerKVSyncStatus.Ready, sequence);
|
||||
}
|
||||
}
|
||||
|
||||
internal PlayerKVStateResult CaptureOrdinaryBatch(out PlayerKVBatch batch)
|
||||
{
|
||||
lock (gate)
|
||||
{
|
||||
batch = null;
|
||||
if (status != PlayerKVSyncStatus.Ready || snapshot.Active != null || snapshot.Unknown != null || snapshot.AuthenticationInvalid) return Blocked();
|
||||
if (throughSequence < 0) return Invalid();
|
||||
var ordered = snapshot.Pending.Values.Where(p => p.Sequence <= throughSequence).OrderBy(p => p.Sequence).ToArray();
|
||||
if (ordered.Length == 0) return new PlayerKVStateResult(PlayerKVChangeError.None, revision: snapshot.Revision);
|
||||
if (status != PlayerKVSyncStatus.Ready || snapshot.AuthenticationInvalid)
|
||||
{
|
||||
return Blocked();
|
||||
}
|
||||
if (snapshot.OrdinaryBuffers[0].Count == 0)
|
||||
{
|
||||
return new PlayerKVStateResult(PlayerKVChangeError.None, revision: snapshot.Revision);
|
||||
}
|
||||
var copy = snapshot.Clone();
|
||||
copy.Active = new PlayerKVBatch(Guid.NewGuid().ToString("N"), ordered.TakeWhile(p => p.Target.Exists == ordered[0].Target.Exists).Take(10));
|
||||
var first = copy.OrdinaryBuffers[0];
|
||||
var captured = new PlayerKVBatch(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, PlayerKVChange>(StringComparer.Ordinal));
|
||||
var result = Commit(copy, PlayerKVSyncStatus.Ready);
|
||||
if (result.Success) batch = snapshot.Active; // Never expose a batch until its write-ahead record is durable.
|
||||
if (result.Success)
|
||||
{
|
||||
batch = captured;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
internal PlayerKVStateResult BeginOrdinaryBatch(out PlayerKVBatch batch)
|
||||
{
|
||||
lock (gate)
|
||||
{
|
||||
batch = null;
|
||||
if (status != PlayerKVSyncStatus.Ready || snapshot.Active != null || snapshot.Unknown != null || snapshot.AuthenticationInvalid)
|
||||
{
|
||||
return Blocked();
|
||||
}
|
||||
if (snapshot.PreparedOrdinaryBatches.Count == 0)
|
||||
{
|
||||
return new PlayerKVStateResult(PlayerKVChangeError.None, revision: snapshot.Revision);
|
||||
}
|
||||
var copy = snapshot.Clone();
|
||||
copy.Active = copy.PreparedOrdinaryBatches[0];
|
||||
copy.PreparedOrdinaryBatches.RemoveAt(0);
|
||||
var result = Commit(copy, PlayerKVSyncStatus.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.
|
||||
public PlayerKVStateResult BeginBatch(long throughSequence, out PlayerKVBatch batch)
|
||||
{
|
||||
return BeginBatch(throughSequence, false, out batch);
|
||||
}
|
||||
|
||||
internal PlayerKVStateResult BeginBatch(long throughSequence, bool legacyOnly, out PlayerKVBatch batch)
|
||||
{
|
||||
lock (gate)
|
||||
{
|
||||
batch = null;
|
||||
if (status != PlayerKVSyncStatus.Ready || snapshot.Active != null || snapshot.Unknown != null || snapshot.AuthenticationInvalid)
|
||||
{
|
||||
return Blocked();
|
||||
}
|
||||
if (throughSequence < 0)
|
||||
{
|
||||
return Invalid();
|
||||
}
|
||||
var source = legacyOnly ? snapshot.LegacyPending.Values : snapshot.LegacyPending.Values.Concat(snapshot.ExplicitDeletes.Values);
|
||||
var ordered = source.Where(item => item.Sequence <= throughSequence).OrderBy(item => item.Sequence).ToArray();
|
||||
if (ordered.Length == 0)
|
||||
{
|
||||
return new PlayerKVStateResult(PlayerKVChangeError.None, revision: snapshot.Revision);
|
||||
}
|
||||
var copy = snapshot.Clone();
|
||||
copy.Active = new PlayerKVBatch(Guid.NewGuid().ToString("N"), ordered.TakeWhile(item => item.Target.Exists == ordered[0].Target.Exists).Take(10));
|
||||
var result = Commit(copy, PlayerKVSyncStatus.Ready);
|
||||
if (result.Success)
|
||||
{
|
||||
batch = snapshot.Active;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
@@ -142,35 +587,204 @@ namespace game.PlayerDataStorage
|
||||
{
|
||||
lock (gate)
|
||||
{
|
||||
if (status != PlayerKVSyncStatus.Ready || snapshot.Active == null || snapshot.Active.Id != batchId) return Blocked();
|
||||
if (status != PlayerKVSyncStatus.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);
|
||||
var current = copy.Pending[item.Key];
|
||||
if (current.Sequence == item.Sequence) copy.Pending.Remove(item.Key);
|
||||
else copy.Pending[item.Key] = current.WithBase(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, PlayerKVSyncStatus.Ready);
|
||||
}
|
||||
}
|
||||
|
||||
// mayHaveCommitted=false is only valid with positive evidence of no dispatch.
|
||||
// A definitely undispatched ordinary batch retires, as the original SDK call did; explicit source references stay.
|
||||
public PlayerKVStateResult FailBatch(string batchId, bool mayHaveCommitted, bool unauthorized = false)
|
||||
{
|
||||
lock (gate)
|
||||
{
|
||||
if (Terminal || snapshot.Active == null || snapshot.Active.Id != batchId) return Blocked();
|
||||
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) copy.Unknown = copy.Active;
|
||||
if (mayHaveCommitted)
|
||||
{
|
||||
if (ordinary)
|
||||
{
|
||||
copy.OrdinaryUnknownBatches.Add(copy.Active);
|
||||
}
|
||||
else
|
||||
{
|
||||
copy.Unknown = copy.Active;
|
||||
}
|
||||
}
|
||||
copy.Active = null;
|
||||
return Commit(copy, mayHaveCommitted ? PlayerKVSyncStatus.UnknownOutcome :
|
||||
copy.AuthenticationInvalid ? PlayerKVSyncStatus.Unauthorized : PlayerKVSyncStatus.ReadFailed);
|
||||
var next = copy.AuthenticationInvalid ? PlayerKVSyncStatus.Unauthorized : ordinary ? PlayerKVSyncStatus.Ready : PlayerKVSyncStatus.ReadFailed;
|
||||
if (copy.Unknown != null)
|
||||
{
|
||||
next = PlayerKVSyncStatus.UnknownOutcome;
|
||||
}
|
||||
return Commit(copy, next);
|
||||
}
|
||||
}
|
||||
|
||||
private static IEnumerable<Dictionary<string, PlayerKVChange>> 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, PlayerKVValue baseline)
|
||||
{
|
||||
foreach (var source in Sources(value))
|
||||
{
|
||||
if (source.TryGetValue(key, out var pending))
|
||||
{
|
||||
source[key] = pending.WithBase(baseline);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
internal PlayerKVStateResult PrepareImmediate(string key, string value, out PlayerKVChange operation)
|
||||
{
|
||||
operation = null;
|
||||
if (!KVSnapshot.ValidKey(key) || value == null)
|
||||
{
|
||||
// Preserve the single-key state admission order and validation result.
|
||||
lock (gate)
|
||||
{
|
||||
return status != PlayerKVSyncStatus.Ready || snapshot.AuthenticationInvalid ? Blocked() : Invalid();
|
||||
}
|
||||
}
|
||||
var result = PrepareImmediate(new Dictionary<string, string> { [key] = value }, out var operations);
|
||||
if (result.Success)
|
||||
{
|
||||
operation = operations[0];
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
internal PlayerKVStateResult PrepareImmediate(IReadOnlyDictionary<string, string> data,
|
||||
out IReadOnlyList<PlayerKVChange> operations)
|
||||
{
|
||||
lock (gate)
|
||||
{
|
||||
operations = null;
|
||||
if (status != PlayerKVSyncStatus.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<PlayerKVChange>();
|
||||
foreach (var pair in data)
|
||||
{
|
||||
copy.ImmediatePreservedKeys.Add(pair.Key);
|
||||
var operation = new PlayerKVChange(pair.Key, copy.NextSequence++,
|
||||
PlayerKVValue.At(copy.Baseline, pair.Key), PlayerKVValue.Present(pair.Value));
|
||||
copy.RemoteOnlyWrites.Add(operation.Sequence, operation);
|
||||
RemoveSources(copy, pair.Key);
|
||||
prepared.Add(operation);
|
||||
}
|
||||
var result = Commit(copy, PlayerKVSyncStatus.Ready, prepared[0].Sequence);
|
||||
if (result.Success)
|
||||
{
|
||||
operations = prepared.AsReadOnly();
|
||||
if (comparisonLocal != null)
|
||||
{
|
||||
comparisonImmediateKeys.UnionWith(data.Keys);
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
internal PlayerKVStateResult CompleteImmediate(PlayerKVChange operation)
|
||||
{
|
||||
return CompleteImmediate(new[] { operation });
|
||||
}
|
||||
|
||||
internal PlayerKVStateResult CompleteImmediate(IReadOnlyList<PlayerKVChange> operations)
|
||||
{
|
||||
lock (gate)
|
||||
{
|
||||
if (status != PlayerKVSyncStatus.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, PlayerKVSyncStatus.Ready);
|
||||
}
|
||||
}
|
||||
|
||||
internal void CancelUndispatchedImmediate(PlayerKVChange operation)
|
||||
{
|
||||
CancelUndispatchedImmediate(new[] { operation });
|
||||
}
|
||||
|
||||
internal void CancelUndispatchedImmediate(IReadOnlyList<PlayerKVChange> 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);
|
||||
}
|
||||
|
||||
bool Terminal => status == PlayerKVSyncStatus.Closed || status == PlayerKVSyncStatus.StorageError;
|
||||
PlayerKVStateResult Commit(KVSnapshot copy, PlayerKVSyncStatus nextStatus, long sequence = 0)
|
||||
{
|
||||
@@ -196,6 +810,7 @@ namespace game.PlayerDataStorage
|
||||
{
|
||||
if (status == PlayerKVSyncStatus.Closed) return;
|
||||
status = PlayerKVSyncStatus.Closed;
|
||||
ClearComparison();
|
||||
store.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user