feat(bridge): protocol 6 — an idempotency key, and champ.boss.killed (Phase 11a)
A command carrying an `idempotencyKey` is now executed at most once: a repeat is answered with the original reply rather than re-run. That is the precondition every world verb in Phase 12 is waiting on, and it is what let `uo.broadcast` stop being un-retryable. The gate sits in BridgeBoot's inbound dispatch, not in each handler, so it covers every kind including ones a later protocol adds. A command with no key behaves exactly as it did before, which leaves the admin screens unchanged. Four rules, each a decision rather than an implementation detail: reserve on receipt (so a handler that defers is covered, answering `bridge.busy` to a repeat in flight); a key that has begun is never released, not even when the handler throws; a replay is stamped with the REPEAT's correlation id, because the sidecar's reqId is fresh per call and replaying the original would hang the retry; and the bound is loud, because an evicted key is the guarantee's one hole. `champ.boss.killed` rides along because a bump costs a release, a bundle and an operator update on every shard. It fires from EventSink.CreatureDeath, detected by type so a boss that popped and died inside one sweep is still reported, and it carries the damage table that exists at the death and nowhere else. overlay.toml protocol = 6, in this commit rather than a later one. Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -133,7 +133,42 @@ namespace Server.Custom.Bridge
|
||||
return;
|
||||
}
|
||||
|
||||
handler(obj);
|
||||
// Protocol 6. A command may carry an `idempotencyKey`, and one that does is executed at
|
||||
// most once: a repeat is answered with the original reply rather than re-run. The gate
|
||||
// is here rather than in each handler so it covers every inbound kind — including the
|
||||
// ones a later protocol adds, which is the half that is easy to forget. A command with
|
||||
// no key behaves exactly as it did before, which is what keeps the admin screens (which
|
||||
// send none) unchanged.
|
||||
var idempotencyKey = BridgeJson.GetString(obj, "idempotencyKey");
|
||||
|
||||
if (idempotencyKey == null)
|
||||
{
|
||||
handler(obj);
|
||||
return;
|
||||
}
|
||||
|
||||
if (BridgeIdempotency.Intercept(idempotencyKey, obj))
|
||||
return; // already answered: a replay of the original reply, or bridge.busy
|
||||
|
||||
string error = null;
|
||||
|
||||
try
|
||||
{
|
||||
handler(obj);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Swallowed deliberately, and only on the keyed path: the key must be closed out
|
||||
// with a definite answer (see BridgeIdempotency's header) rather than left in
|
||||
// flight by an exception unwinding past Finish. Unkeyed commands still throw the
|
||||
// way they always have.
|
||||
error = ex.Message;
|
||||
Console.WriteLine("[Bridge] handler for '{0}' threw: {1}", kind, ex);
|
||||
}
|
||||
finally
|
||||
{
|
||||
BridgeIdempotency.Finish(idempotencyKey, error);
|
||||
}
|
||||
}
|
||||
|
||||
private static void OnPing(Dictionary<string, object> o)
|
||||
@@ -215,6 +250,7 @@ namespace Server.Custom.Bridge
|
||||
e.Mobile.SendMessage("Bridge: {0}", BridgeMarket.Status());
|
||||
e.Mobile.SendMessage("Bridge: {0}", BridgePages.Status());
|
||||
e.Mobile.SendMessage("Bridge: {0}", BridgeRuleset.Status());
|
||||
e.Mobile.SendMessage("Bridge: {0}", BridgeIdempotency.Status());
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -41,7 +41,17 @@ namespace Server.Custom.Bridge
|
||||
// Item and Mobile serials occupy disjoint ranges, so one map safely spans all three families.
|
||||
private static readonly Dictionary<Serial, string> _last = new Dictionary<Serial, string>();
|
||||
|
||||
private static long _sweeps, _emitted, _removed;
|
||||
// Protocol 6. Which spawn a live champion belongs to, refreshed by the sweep. The kill itself
|
||||
// is detected by TYPE (see OnCreatureDeath), so this map only ever supplies CONTEXT — which
|
||||
// altar, at what level. A boss that popped and died inside one sweep interval is still
|
||||
// reported; it simply arrives without its spawn.
|
||||
private static readonly Dictionary<Serial, Serial> _bossOf = new Dictionary<Serial, Serial>();
|
||||
|
||||
// How many damage entries a kill reports. Deep enough that a real champion fight's meaningful
|
||||
// contributors are all present, shallow enough that the frame stays one line on the wire.
|
||||
private const int MaxDamagers = 20;
|
||||
|
||||
private static long _sweeps, _emitted, _removed, _bossKills;
|
||||
|
||||
public static void Initialize()
|
||||
{
|
||||
@@ -56,12 +66,24 @@ namespace Server.Custom.Bridge
|
||||
// Re-emit the full board whenever the sidecar (re)connects, so a sidecar that restarted
|
||||
// independently of the shard rebuilds its state within one sweep.
|
||||
BridgeLink.Connected_Core += OnConnected;
|
||||
|
||||
// Protocol 6. A boss defeat was previously only INFERABLE — champ.update going bossUp
|
||||
// true then false, correlated against a mob.killed nearby — and that inference is both
|
||||
// fragile and silent about who did the work. It is a real moment in a shard's week and
|
||||
// an event's phase condition wants to name it, so it becomes a kind of its own.
|
||||
EventSink.CreatureDeath += OnCreatureDeath;
|
||||
|
||||
Rearm();
|
||||
}
|
||||
|
||||
private static void OnConnected()
|
||||
{
|
||||
_last.Clear();
|
||||
|
||||
// _bossOf is deliberately NOT cleared. It is a fact about the world, not a diff cache:
|
||||
// dropping it on a sidecar reconnect would lose the spawn attribution for a boss that is
|
||||
// up right now, and it refills from the sweep only if that boss's record happens to
|
||||
// change again before it dies.
|
||||
}
|
||||
|
||||
/// <summary>Stops and recreates the timer from current config. Called by `[bridge reload`.</summary>
|
||||
@@ -82,8 +104,160 @@ namespace Server.Custom.Bridge
|
||||
|
||||
public static string Status()
|
||||
{
|
||||
return String.Format("champs(sweeps={0} emitted={1} removed={2} tracked={3})",
|
||||
_sweeps, _emitted, _removed, _last.Count);
|
||||
return String.Format("champs(sweeps={0} emitted={1} removed={2} tracked={3} bossKills={4} bossesUp={5})",
|
||||
_sweeps, _emitted, _removed, _last.Count, _bossKills, _bossOf.Count);
|
||||
}
|
||||
|
||||
// ---- champ.boss.killed (Protocol 6) ----
|
||||
|
||||
/// <summary>
|
||||
/// Fires for every creature death on the shard, so the first thing it does is decide
|
||||
/// this is not one. Detection is by TYPE — <c>BaseChampion</c>, which
|
||||
/// <c>BaseSeaChampion</c> derives from, so one check covers both families — with the
|
||||
/// sweep's map used only to name the altar. A boss that popped and died between two
|
||||
/// sweeps is therefore still reported; it simply arrives without a spawn.
|
||||
///
|
||||
/// The damage table is read here and nowhere else, because it exists here and nowhere
|
||||
/// else: ServUO discards a creature's damage entries with the creature, and the shard is
|
||||
/// the only party that ever sees them. Entries are reported whether or not ServUO
|
||||
/// considers them expired — expiry governs LOOTING RIGHTS, and someone who fought the
|
||||
/// first two thirds of a champion fight and then died took part in it regardless of what
|
||||
/// they are owed from the corpse.
|
||||
/// </summary>
|
||||
private static void OnCreatureDeath(CreatureDeathEventArgs e)
|
||||
{
|
||||
try
|
||||
{
|
||||
var boss = e.Creature;
|
||||
|
||||
if (boss == null)
|
||||
return;
|
||||
|
||||
Serial spawnSerial;
|
||||
bool attributed = _bossOf.TryGetValue(boss.Serial, out spawnSerial);
|
||||
|
||||
if (!(boss is BaseChampion) && !attributed)
|
||||
return;
|
||||
|
||||
_bossOf.Remove(boss.Serial);
|
||||
_bossKills++;
|
||||
|
||||
var spawn = attributed ? World.FindItem(spawnSerial) as ChampionSpawn : null;
|
||||
var name = String.IsNullOrEmpty(boss.Name) ? boss.GetType().Name : boss.Name;
|
||||
|
||||
var sb = BridgeJson.Begin("champ.boss.killed")
|
||||
.Str("category", boss is BaseSeaChampion ? "sea" : "champion")
|
||||
.Ser("bossSerial", boss.Serial)
|
||||
.Str("boss", name)
|
||||
.Str("bossType", boss.GetType().Name)
|
||||
.Str("map", boss.Map == null ? null : boss.Map.Name)
|
||||
.Num("x", boss.X).Num("y", boss.Y).Num("z", boss.Z);
|
||||
|
||||
// The altar's own record, when the kill could be attributed to one. `serial` is the
|
||||
// SPAWN here, matching champ.update, so a consumer can join the two without a rule
|
||||
// about which of two serials on the frame means what.
|
||||
if (spawn != null)
|
||||
{
|
||||
sb.Ser("serial", spawn.Serial)
|
||||
.Str("type", spawn.Type.ToString())
|
||||
.Num("level", spawn.Level);
|
||||
}
|
||||
|
||||
// A named region is what a phase condition can actually match on ("the boss in
|
||||
// Yew"); coordinates are not. Emitted alongside the coordinates rather than
|
||||
// instead, because large stretches of the map belong to no named region at all.
|
||||
//
|
||||
// **The innermost region here is ANONYMOUS, and the rig is the only thing that was
|
||||
// ever going to say so.** A champion killed in the middle of Britain produced a
|
||||
// frame with no region at all, because an active `ChampionSpawn` registers a
|
||||
// `ChampionSpawnRegion` over its own spawn area — constructed with a null name and
|
||||
// with the town region as its PARENT (`ChampionSpawn.cs`, its constructor). So the
|
||||
// most specific region containing a champion boss is, by construction, the one
|
||||
// region on the map guaranteed to have no name.
|
||||
//
|
||||
// It also explains why this looked fine for twenty seconds: region registration is
|
||||
// deferred, so a lookup immediately after the altar is placed still answers
|
||||
// "Britain" and one at the kill does not. A first read at spawn time would have
|
||||
// confirmed a bug into the design.
|
||||
//
|
||||
// Walking to the nearest NAMED ancestor is the general answer rather than a special
|
||||
// case for champions: a house region, a dungeon sub-region and a guarded-zone
|
||||
// overlay are all anonymous children of somewhere a player would name.
|
||||
var region = NamedRegionAt(boss.Location, boss.Map);
|
||||
|
||||
if (region != null)
|
||||
sb.Str("region", region);
|
||||
|
||||
if (e.Killer != null)
|
||||
sb.Actor("killer", e.Killer);
|
||||
|
||||
sb.Damagers("damagers", TopDamagers(boss), MaxDamagers);
|
||||
|
||||
BridgeLink.Emit(sb.End());
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// A death handler must never be the thing that breaks a death.
|
||||
Console.WriteLine("[Bridge] champ.boss.killed threw: {0}", ex.Message);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Player damage against this creature, highest first. Totals are summed per damager
|
||||
/// rather than trusted to be one entry each: ServUO's own registration folds repeat
|
||||
/// damage into an existing entry, but an entry that expired and was re-created leaves
|
||||
/// two, and a table that listed the same player twice would be read as two participants.
|
||||
/// </summary>
|
||||
/// <summary>
|
||||
/// The nearest NAMED region containing a point, walking outward from the most specific
|
||||
/// one, or null when nothing on the way out has a name.
|
||||
///
|
||||
/// Null rather than "" so the caller can leave the field off the frame entirely: a
|
||||
/// consumer reading `region: ""` cannot tell "outdoors, nowhere in particular" from
|
||||
/// "somewhere, but the shard would not say", and only one of those is true here.
|
||||
///
|
||||
/// The map's own default region terminates the walk with its parentless empty name, so
|
||||
/// a point in open countryside answers null without a special case.
|
||||
/// </summary>
|
||||
private static string NamedRegionAt(Point3D p, Map map)
|
||||
{
|
||||
if (map == null)
|
||||
return null;
|
||||
|
||||
for (var region = Region.Find(p, map); region != null; region = region.Parent)
|
||||
{
|
||||
if (!String.IsNullOrEmpty(region.Name))
|
||||
return region.Name;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
private static List<KeyValuePair<Mobile, int>> TopDamagers(Mobile boss)
|
||||
{
|
||||
var totals = new Dictionary<Mobile, int>();
|
||||
|
||||
var entries = boss.DamageEntries;
|
||||
|
||||
if (entries != null)
|
||||
{
|
||||
for (int i = 0; i < entries.Count; i++)
|
||||
{
|
||||
var de = entries[i];
|
||||
|
||||
if (de == null || de.Damager == null || de.Damager.Deleted || !de.Damager.Player)
|
||||
continue;
|
||||
|
||||
int running;
|
||||
totals.TryGetValue(de.Damager, out running);
|
||||
totals[de.Damager] = running + de.DamageGiven;
|
||||
}
|
||||
}
|
||||
|
||||
var ranked = totals.ToList();
|
||||
ranked.Sort((a, b) => b.Value.CompareTo(a.Value));
|
||||
|
||||
return ranked;
|
||||
}
|
||||
|
||||
/// <summary>Runs one sweep now. Wired into `[bridge sweepnow`.</summary>
|
||||
@@ -107,6 +281,15 @@ namespace Server.Custom.Bridge
|
||||
{
|
||||
if (s.Deleted)
|
||||
continue;
|
||||
|
||||
// Protocol 6. Remember which altar a live champion belongs to so its death can
|
||||
// name one. Recorded here rather than looked up at death because the lookup
|
||||
// would be a scan of World.Items on every creature death on the shard.
|
||||
var champion = s.Champion;
|
||||
|
||||
if (champion != null && !champion.Deleted)
|
||||
_bossOf[champion.Serial] = s.Serial;
|
||||
|
||||
Track(seen, s.Serial, SigChampion(s), WriteChampion(s));
|
||||
}
|
||||
|
||||
@@ -133,6 +316,19 @@ namespace Server.Custom.Bridge
|
||||
BridgeLink.Emit(BridgeJson.Begin("champ.remove").Ser("serial", serial).End());
|
||||
_removed++;
|
||||
}
|
||||
|
||||
// A defeated champion's attribution is consumed by OnCreatureDeath, but one deleted
|
||||
// by a GM or lost to a world reload never dies, so the map is swept too. Cheap: it
|
||||
// holds at most one entry per altar with a boss currently up.
|
||||
if (_bossOf.Count > 0)
|
||||
{
|
||||
var vanished = _bossOf.Keys
|
||||
.Where(k => { var m = World.FindMobile(k); return m == null || m.Deleted; })
|
||||
.ToList();
|
||||
|
||||
foreach (var k in vanished)
|
||||
_bossOf.Remove(k);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
|
||||
439
overlay/Scripts/Custom/Bridge/BridgeIdempotency.cs
Normal file
439
overlay/Scripts/Custom/Bridge/BridgeIdempotency.cs
Normal file
@@ -0,0 +1,439 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace Server.Custom.Bridge
|
||||
{
|
||||
/// <summary>
|
||||
/// Protocol 6. Makes a repeated command safe.
|
||||
///
|
||||
/// The website's event runner retries a step that did not come back, and until now a command
|
||||
/// whose acknowledgement was lost was indistinguishable from one that never applied. There
|
||||
/// was no way to tell the difference from either end, so every world-writing verb had to be
|
||||
/// declared un-retryable — a lost announcement being cheaper than a doubled one. That is not
|
||||
/// a position you can hold once an event can spawn creatures or lease a config value.
|
||||
///
|
||||
/// So a command may now carry an `idempotencyKey`, and the shard promises: **a key is
|
||||
/// executed at most once.** A repeat is never re-run. It is answered with the original
|
||||
/// reply — the same acknowledgement the caller lost — under the repeat's own correlation id.
|
||||
///
|
||||
/// ── Reserve on receipt, not on completion ──────────────────────────────────────────────
|
||||
///
|
||||
/// The key is recorded BEFORE the handler is dispatched, not after it returns. A handler that
|
||||
/// finishes inside its own inbound call can never see a repeat (the Core thread processes one
|
||||
/// line at a time), but a handler that defers — a lease that arms a timer, a spawn that
|
||||
/// waits for a save — completes long after `OnInboundLine` has returned, and that is exactly
|
||||
/// the window a lost ack opens. Reserving late would leave it uncovered.
|
||||
///
|
||||
/// A repeat of a key that is still in flight is answered `bridge.busy`: it runs nothing and
|
||||
/// tells the caller to come back. `bridge.busy` is deliberately not an error — the work is
|
||||
/// happening, and the module classifies it retryable.
|
||||
///
|
||||
/// ── A key that has begun is never released ────────────────────────────────────────────
|
||||
///
|
||||
/// Not even when the handler throws. Releasing it would let a retry re-run a command that may
|
||||
/// have applied half of itself, which is precisely the failure this file exists to prevent.
|
||||
/// A handler that throws stores a `bridge.error` reply instead, so the retry gets a definite
|
||||
/// answer and the step fails once rather than looping.
|
||||
///
|
||||
/// ── The one hole, and why it is loud ──────────────────────────────────────────────────
|
||||
///
|
||||
/// The set is bounded, so an evicted key's repeat WOULD be applied a second time. The bounds
|
||||
/// are chosen to put that far outside reach — an hour, against core's fifteen-minute step
|
||||
/// lease — and an eviction that drops a key which had not yet expired prints a warning naming
|
||||
/// the count. If the guarantee is ever actually breached, an operator sees it here rather
|
||||
/// than discovering a doubled spawn in the world.
|
||||
/// </summary>
|
||||
public static class BridgeIdempotency
|
||||
{
|
||||
/// <summary>
|
||||
/// How long a key is remembered. Core's step lease is 15 minutes and its retry backoff
|
||||
/// is bounded well inside that, so an hour is not a tuned number — it is a margin wide
|
||||
/// enough that expiry should never be the thing that ends a key's life.
|
||||
/// </summary>
|
||||
private static readonly TimeSpan Ttl = TimeSpan.FromHours(1.0);
|
||||
|
||||
/// <summary>
|
||||
/// Hard bound on remembered keys, in the same spirit as BridgeLink's outbound queue cap:
|
||||
/// the Core thread never holds an unbounded collection. At command rates this is days of
|
||||
/// traffic, so reaching it means something is wrong — hence the warning on eviction.
|
||||
/// </summary>
|
||||
private const int Cap = 4096;
|
||||
|
||||
/// <summary>
|
||||
/// The correlation fields the sidecar routes replies on, in the order `rpc.rs` tries
|
||||
/// them. A command carries exactly one; the reply echoes it. A replay must be stamped
|
||||
/// with the REPEAT's value, not the original's — the sidecar's `reqId` is a fresh
|
||||
/// per-process counter, so the retry is waiting on an id the first attempt never used.
|
||||
/// </summary>
|
||||
private static readonly string[] CorrFields = { "reqId", "code", "id" };
|
||||
|
||||
private sealed class Entry
|
||||
{
|
||||
public DateTime Reserved; // when the key was first seen
|
||||
public bool Done; // the handler has finished (successfully or not)
|
||||
public string Reply; // the correlated reply line, verbatim; null if there was none
|
||||
public string Corr; // the correlation value the original reply carries
|
||||
public string CorrField; // which of CorrFields that value sits in
|
||||
public string Kind; // for diagnostics only
|
||||
}
|
||||
|
||||
private static readonly Dictionary<string, Entry> _byKey =
|
||||
new Dictionary<string, Entry>(StringComparer.Ordinal);
|
||||
|
||||
// Insertion order, so the cap evicts oldest-first without sorting the dictionary.
|
||||
private static readonly Queue<string> _order = new Queue<string>();
|
||||
|
||||
// ---- capture state; Core thread only, one keyed command at a time ----
|
||||
|
||||
private static Entry _open;
|
||||
private static string _openCorr;
|
||||
private static string _openCorrField;
|
||||
|
||||
private static long _seen, _replayed, _busy, _evicted, _uncorrelated;
|
||||
|
||||
/// <summary>
|
||||
/// True while a keyed command's handler is running. BridgeLink.Emit checks this on every
|
||||
/// emit, so it is a plain bool read rather than anything that costs the sweep path.
|
||||
/// </summary>
|
||||
public static bool Capturing
|
||||
{
|
||||
get { return _open != null; }
|
||||
}
|
||||
|
||||
public static string Status()
|
||||
{
|
||||
return String.Format(
|
||||
"idem(keys={0} seen={1} replayed={2} busy={3} evicted={4} uncorrelated={5})",
|
||||
_byKey.Count, _seen, _replayed, _busy, _evicted, _uncorrelated);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Called by BridgeBoot for every inbound command that carries an `idempotencyKey`,
|
||||
/// before the handler runs.
|
||||
///
|
||||
/// Returns TRUE when the command must not be executed — this call has already emitted the
|
||||
/// answer (a replay of the original reply, or `bridge.busy`). Returns FALSE when the key
|
||||
/// is new: the key is now reserved and capture is open, and the caller MUST pair this
|
||||
/// with <see cref="Finish"/> in a finally.
|
||||
/// </summary>
|
||||
public static bool Intercept(string key, Dictionary<string, object> command)
|
||||
{
|
||||
_seen++;
|
||||
Sweep();
|
||||
|
||||
string corrField = null;
|
||||
string corr = null;
|
||||
|
||||
for (int i = 0; i < CorrFields.Length; i++)
|
||||
{
|
||||
var v = BridgeJson.GetString(command, CorrFields[i]);
|
||||
|
||||
if (v != null)
|
||||
{
|
||||
corrField = CorrFields[i];
|
||||
corr = v;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
Entry prior;
|
||||
|
||||
if (_byKey.TryGetValue(key, out prior))
|
||||
{
|
||||
if (prior.Done)
|
||||
Replay(key, prior, corrField, corr);
|
||||
else
|
||||
Busy(key, prior, corrField, corr);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
var entry = new Entry
|
||||
{
|
||||
Reserved = DateTime.UtcNow,
|
||||
Done = false,
|
||||
Kind = BridgeJson.GetString(command, "kind"),
|
||||
};
|
||||
|
||||
Remember(key, entry);
|
||||
|
||||
_open = entry;
|
||||
_openCorr = corr;
|
||||
_openCorrField = corrField;
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Called by BridgeBoot in a finally, once the handler has returned. Closes capture and
|
||||
/// marks the key done. `error` is non-null when the handler threw.
|
||||
///
|
||||
/// A handler that deferred its work calls <see cref="Hold"/> first; this then leaves the
|
||||
/// key reserved and in flight, and the handler completes it later.
|
||||
/// </summary>
|
||||
public static void Finish(string key, string error)
|
||||
{
|
||||
var entry = _open;
|
||||
|
||||
_open = null;
|
||||
var corr = _openCorr;
|
||||
var corrField = _openCorrField;
|
||||
_openCorr = null;
|
||||
_openCorrField = null;
|
||||
|
||||
if (entry == null || entry.Done)
|
||||
return; // Hold() released it to its own completion, or there was nothing open
|
||||
|
||||
if (error != null)
|
||||
{
|
||||
// The handler threw. The key stays claimed — see the class header — and the stored
|
||||
// answer is the failure, so the retry ends the step instead of re-running a command
|
||||
// that may have applied part of itself.
|
||||
var sb = BridgeJson.Begin("bridge.error");
|
||||
|
||||
if (corrField != null)
|
||||
sb.Str(corrField, corr);
|
||||
|
||||
sb.Str("reason", "handler threw: " + error)
|
||||
.Str("idempotencyKey", key);
|
||||
|
||||
entry.Reply = sb.End();
|
||||
entry.Corr = corr;
|
||||
entry.CorrField = corrField;
|
||||
entry.Done = true;
|
||||
|
||||
Console.WriteLine("[Bridge] idempotency: {0} threw under key {1}; the retry will be answered with the failure",
|
||||
entry.Kind, key);
|
||||
return;
|
||||
}
|
||||
|
||||
if (entry.Reply == null)
|
||||
{
|
||||
// Nothing the sidecar could have correlated was emitted. That is a defect in the
|
||||
// handler rather than a state to model: the FIRST attempt has already timed out at
|
||||
// the sidecar, and the retry would time out identically forever. Store a definite
|
||||
// answer so the retry terminates, and say so.
|
||||
_uncorrelated++;
|
||||
|
||||
var sb = BridgeJson.Begin("bridge.error");
|
||||
|
||||
if (corrField != null)
|
||||
sb.Str(corrField, corr);
|
||||
|
||||
sb.Str("reason", "the original command produced no correlated reply")
|
||||
.Str("idempotencyKey", key);
|
||||
|
||||
entry.Reply = sb.End();
|
||||
entry.Corr = corr;
|
||||
entry.CorrField = corrField;
|
||||
|
||||
Console.WriteLine("[Bridge] idempotency: {0} under key {1} emitted no reply the sidecar could correlate",
|
||||
entry.Kind, key);
|
||||
}
|
||||
|
||||
entry.Done = true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// For a handler that finishes AFTER its inbound call returns. It keeps the key reserved
|
||||
/// (so a repeat is answered `bridge.busy` rather than executed) and takes on the duty of
|
||||
/// calling <see cref="Complete"/> with the reply it eventually emits.
|
||||
///
|
||||
/// Nothing in protocol 6 defers yet. The door exists because the leases and world verbs
|
||||
/// that follow do, and because a deferred handler that had no way to hold its key would
|
||||
/// quietly be the one place the guarantee did not hold.
|
||||
/// </summary>
|
||||
public static void Hold(string key)
|
||||
{
|
||||
var entry = _open;
|
||||
|
||||
if (entry == null)
|
||||
return;
|
||||
|
||||
// Close capture without marking done: the key stays in flight until Complete.
|
||||
_open = null;
|
||||
_openCorr = null;
|
||||
_openCorrField = null;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Completes a key a handler previously held. `replyLine` is the line the handler emits
|
||||
/// as its answer; it is stored so a later repeat replays it.
|
||||
/// </summary>
|
||||
public static void Complete(string key, string replyLine)
|
||||
{
|
||||
Entry entry;
|
||||
|
||||
if (key == null || !_byKey.TryGetValue(key, out entry) || entry.Done)
|
||||
return;
|
||||
|
||||
var parsed = replyLine == null ? null : BridgeJson.Parse(replyLine);
|
||||
|
||||
if (parsed != null)
|
||||
{
|
||||
for (int i = 0; i < CorrFields.Length; i++)
|
||||
{
|
||||
var v = BridgeJson.GetString(parsed, CorrFields[i]);
|
||||
|
||||
if (v != null)
|
||||
{
|
||||
entry.CorrField = CorrFields[i];
|
||||
entry.Corr = v;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
entry.Reply = replyLine;
|
||||
entry.Done = true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Every line a keyed handler emits passes through here. Only the one the sidecar would
|
||||
/// correlate with THIS command is kept: an `admin.audit` broadcast that happens to be
|
||||
/// emitted alongside the reply is a fact about the world and must not be replayed, while
|
||||
/// the reply is an answer to a caller and must be.
|
||||
/// </summary>
|
||||
public static void Observe(string line)
|
||||
{
|
||||
var entry = _open;
|
||||
|
||||
if (entry == null || line == null || _openCorrField == null || _openCorr == null)
|
||||
return;
|
||||
|
||||
// Cheap reject before parsing: the correlation value is a string field on the reply, so
|
||||
// if it does not appear in the line at all this cannot be the reply.
|
||||
if (line.IndexOf(_openCorr, StringComparison.Ordinal) < 0)
|
||||
return;
|
||||
|
||||
var parsed = BridgeJson.Parse(line);
|
||||
|
||||
if (parsed == null)
|
||||
return;
|
||||
|
||||
if (!String.Equals(BridgeJson.GetString(parsed, _openCorrField), _openCorr, StringComparison.Ordinal))
|
||||
return;
|
||||
|
||||
entry.Reply = line;
|
||||
entry.Corr = _openCorr;
|
||||
entry.CorrField = _openCorrField;
|
||||
}
|
||||
|
||||
// ---- internals ----
|
||||
|
||||
private static void Replay(string key, Entry prior, string corrField, string corr)
|
||||
{
|
||||
_replayed++;
|
||||
|
||||
// A repeat with no correlation field is nobody's outstanding call. Re-emitting the
|
||||
// original reply would put a stale answer on the event feed, where a subscriber would
|
||||
// read it as a fresh one, so the repeat is absorbed silently instead.
|
||||
if (corrField == null || corr == null)
|
||||
{
|
||||
Console.WriteLine("[Bridge] idempotency: absorbed an uncorrelated repeat of key {0} ({1})",
|
||||
key, prior.Kind);
|
||||
return;
|
||||
}
|
||||
|
||||
// Stamp the repeat's correlation id over the original's. The sidecar is waiting on the
|
||||
// id IT sent this time; replaying the first attempt's id would leave the call hanging
|
||||
// until the reply timeout, which is the very failure being answered.
|
||||
string line = null;
|
||||
|
||||
if (prior.Reply != null && String.Equals(corrField, prior.CorrField, StringComparison.Ordinal))
|
||||
line = BridgeJson.RewriteStringField(prior.Reply, corrField, corr);
|
||||
|
||||
if (line == null)
|
||||
{
|
||||
// Either the original produced no reply to replay, or the repeat correlates on a
|
||||
// different field than the original did. Nothing sensible can be replayed under an
|
||||
// id the caller is not waiting on, so answer plainly rather than hang the call.
|
||||
BridgeLink.Emit(BridgeJson.Begin("bridge.error")
|
||||
.Str(corrField, corr)
|
||||
.Str("reason", "the original reply for this idempotency key cannot be replayed")
|
||||
.Str("idempotencyKey", key)
|
||||
.End());
|
||||
return;
|
||||
}
|
||||
|
||||
line = BridgeJson.WithTrueFlag(line, "replayed");
|
||||
|
||||
Console.WriteLine("[Bridge] idempotency: replaying the original reply for key {0} ({1})",
|
||||
key, prior.Kind);
|
||||
|
||||
BridgeLink.Emit(line);
|
||||
}
|
||||
|
||||
private static void Busy(string key, Entry prior, string corrField, string corr)
|
||||
{
|
||||
_busy++;
|
||||
|
||||
var sb = BridgeJson.Begin("bridge.busy");
|
||||
|
||||
if (corrField != null)
|
||||
sb.Str(corrField, corr);
|
||||
|
||||
sb.Str("idempotencyKey", key)
|
||||
.Str("kind", prior.Kind)
|
||||
.Str("reason", "a command with this idempotency key is still in flight");
|
||||
|
||||
BridgeLink.Emit(sb.End());
|
||||
}
|
||||
|
||||
private static void Remember(string key, Entry entry)
|
||||
{
|
||||
_byKey[key] = entry;
|
||||
_order.Enqueue(key);
|
||||
|
||||
while (_order.Count > Cap)
|
||||
{
|
||||
var oldest = _order.Dequeue();
|
||||
|
||||
Entry dropped;
|
||||
|
||||
if (!_byKey.TryGetValue(oldest, out dropped))
|
||||
continue;
|
||||
|
||||
_byKey.Remove(oldest);
|
||||
|
||||
// Expired keys leave silently; they are supposed to. A key evicted while still
|
||||
// inside its TTL is the guarantee's one hole, so it never leaves quietly.
|
||||
if (DateTime.UtcNow - dropped.Reserved < Ttl)
|
||||
{
|
||||
_evicted++;
|
||||
Console.WriteLine(
|
||||
"[Bridge] idempotency: evicted key {0} ({1}) while still live — the cap of {2} was reached, so a repeat of it WOULD be applied again ({3} so far)",
|
||||
oldest, dropped.Kind, Cap, _evicted);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Drops keys past their TTL. Runs on the command path, which is human-rate.</summary>
|
||||
private static void Sweep()
|
||||
{
|
||||
if (_order.Count == 0)
|
||||
return;
|
||||
|
||||
var cutoff = DateTime.UtcNow - Ttl;
|
||||
|
||||
while (_order.Count > 0)
|
||||
{
|
||||
var oldest = _order.Peek();
|
||||
|
||||
Entry entry;
|
||||
|
||||
if (!_byKey.TryGetValue(oldest, out entry))
|
||||
{
|
||||
_order.Dequeue();
|
||||
continue;
|
||||
}
|
||||
|
||||
if (entry.Reserved > cutoff)
|
||||
return; // insertion-ordered, so nothing behind this is older
|
||||
|
||||
_order.Dequeue();
|
||||
_byKey.Remove(oldest);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -139,6 +139,53 @@ namespace Server.Custom.Bridge
|
||||
return sb;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A named array of actor objects each carrying a damage total — a boss kill's damage
|
||||
/// table (Protocol 6), and the first actor array whose entries are ranked rather than
|
||||
/// merely listed.
|
||||
///
|
||||
/// The pairs are written in the order given, so the CALLER owns the sort. That is
|
||||
/// deliberate: "the top damagers" is a judgement about a fight, and the shard's job is
|
||||
/// to report the numbers it holds rather than to decide what counts as a contribution.
|
||||
///
|
||||
/// Each entry is the standard actor object plus `damage`, which means it carries `acct`
|
||||
/// and `webId` and is therefore governed by the website's locked-field rule exactly as
|
||||
/// every other actor is. A shard that considers the whole table too revealing hides it
|
||||
/// with one field rule rather than by dropping the kind.
|
||||
/// </summary>
|
||||
public static StringBuilder Damagers(
|
||||
this StringBuilder sb, string name, IList<KeyValuePair<Mobile, int>> pairs, int count)
|
||||
{
|
||||
sb.Append(",\"").Append(name).Append("\":[");
|
||||
|
||||
if (pairs != null)
|
||||
{
|
||||
var end = Math.Min(count, pairs.Count);
|
||||
bool first = true;
|
||||
|
||||
for (int i = 0; i < end; i++)
|
||||
{
|
||||
var m = pairs[i].Key;
|
||||
|
||||
if (m == null)
|
||||
continue;
|
||||
|
||||
if (!first)
|
||||
sb.Append(',');
|
||||
|
||||
sb.Append('{');
|
||||
WriteActorFields(sb, m);
|
||||
sb.Append(",\"damage\":").Append(pairs[i].Value);
|
||||
sb.Append('}');
|
||||
|
||||
first = false;
|
||||
}
|
||||
}
|
||||
|
||||
sb.Append(']');
|
||||
return sb;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A roster member: the standard actor object plus the member's rank in their guild.
|
||||
///
|
||||
@@ -289,6 +336,75 @@ namespace Server.Custom.Bridge
|
||||
sb.Append('"');
|
||||
}
|
||||
|
||||
// ---- rewriting an already-built line (protocol 6) ----
|
||||
//
|
||||
// BridgeIdempotency replays a stored reply under the REPEAT's correlation id. It could
|
||||
// parse the line, edit the dictionary and re-serialize, but a round trip through
|
||||
// JavaScriptSerializer would silently renormalise every number and string in a reply this
|
||||
// file went to the trouble of writing by hand. These two edit the text instead, so a
|
||||
// replayed reply is byte-for-byte the original apart from the field that had to change.
|
||||
|
||||
/// <summary>
|
||||
/// Replaces the value of a top-level STRING field, honouring backslash escapes when
|
||||
/// finding the value's end. Returns null if the field is not present as a string —
|
||||
/// never a half-rewritten line.
|
||||
/// </summary>
|
||||
public static string RewriteStringField(string line, string name, string value)
|
||||
{
|
||||
if (line == null || name == null || value == null)
|
||||
return null;
|
||||
|
||||
// The leading comma is part of the needle: every top-level field is written by Str()
|
||||
// after Begin() has already emitted `t` and `kind`, so a real one always has one. It
|
||||
// is the cheapest thing that stops the search matching the same text inside a value.
|
||||
var needle = ",\"" + name + "\":\"";
|
||||
int at = line.IndexOf(needle, StringComparison.Ordinal);
|
||||
|
||||
if (at < 0)
|
||||
return null;
|
||||
|
||||
int valueStart = at + needle.Length;
|
||||
int i = valueStart;
|
||||
|
||||
while (i < line.Length)
|
||||
{
|
||||
char c = line[i];
|
||||
|
||||
if (c == '\\')
|
||||
{
|
||||
i += 2; // an escape consumes the next character, whatever it is
|
||||
continue;
|
||||
}
|
||||
|
||||
if (c == '"')
|
||||
break;
|
||||
|
||||
i++;
|
||||
}
|
||||
|
||||
if (i >= line.Length)
|
||||
return null; // unterminated: refuse rather than guess
|
||||
|
||||
var sb = new StringBuilder(line.Length + value.Length);
|
||||
sb.Append(line, 0, valueStart - 1); // up to and excluding the opening quote
|
||||
Escape(sb, value);
|
||||
sb.Append(line, i + 1, line.Length - i - 1);
|
||||
|
||||
return sb.ToString();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Appends `"name":true` to an already-closed object. Returns the line unchanged if it
|
||||
/// is not one, so a malformed reply is passed through rather than corrupted further.
|
||||
/// </summary>
|
||||
public static string WithTrueFlag(string line, string name)
|
||||
{
|
||||
if (String.IsNullOrEmpty(line) || line[line.Length - 1] != '}')
|
||||
return line;
|
||||
|
||||
return line.Substring(0, line.Length - 1) + ",\"" + name + "\":true}";
|
||||
}
|
||||
|
||||
// ---- inbound ----
|
||||
|
||||
/// <summary>
|
||||
|
||||
@@ -107,7 +107,17 @@ namespace Server.Custom.Bridge
|
||||
/// </summary>
|
||||
public static void Emit(string line)
|
||||
{
|
||||
if (!_running || line == null)
|
||||
if (line == null)
|
||||
return;
|
||||
|
||||
// Protocol 6. While a keyed command's handler runs — Core thread, one at a time — every
|
||||
// line it emits is offered to the recent-key store so the correlated reply can be
|
||||
// replayed to a retry later. Deliberately BEFORE the `_running` check: a reply the link
|
||||
// was too dead to deliver is precisely the one a retry will come back for.
|
||||
if (BridgeIdempotency.Capturing)
|
||||
BridgeIdempotency.Observe(line);
|
||||
|
||||
if (!_running)
|
||||
return;
|
||||
|
||||
// Drop-oldest. Bound first, then enqueue, so the queue can transiently sit one over
|
||||
|
||||
Reference in New Issue
Block a user