servuo-plugins/overlay/Scripts/Custom/Bridge/BridgeIdempotency.cs

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