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

using System;
using System.Collections.Concurrent;
using System.IO;
using System.Net.Sockets;
using System.Text;
using System.Threading;

namespace Server.Custom.Bridge
{
    /// <summary>
    ///     The loopback link to the Rust sidecar. Newline-delimited JSON, bidirectional.
    ///
    ///     Threading contract, which the whole bridge depends on:
    ///
    ///       * <see cref="Emit"/> is called from the Core thread. It formats nothing, blocks on
    ///         nothing, and touches no socket. It enqueues and returns. A slow, wedged, or absent
    ///         sidecar cannot stall the shard.
    ///       * One link thread owns the socket. It connects, drains the queue, and reconnects with
    ///         backoff. A single writer keeps event ordering intact.
    ///       * A reader thread parses inbound lines and hands each to the Core thread via
    ///         Timer.DelayCall. The reader never touches World, Mobile, Item, or Account.
    ///
    ///     The outbound queue is bounded. On overflow the oldest record is dropped and counted,
    ///     because telemetry is worth less than the shard's memory.
    /// </summary>
    public static class BridgeLink
    {
        private static readonly ConcurrentQueue<string> _outbound = new ConcurrentQueue<string>();
        private static readonly AutoResetEvent _wake = new AutoResetEvent(false);

        private static Thread _link;
        private static volatile bool _running;
        private static volatile bool _connected;

        // Set when the peer goes away, so the writer stops trying.
        private static volatile bool _dead;

        // Incremented per connection attempt. A reader from a previous connection must not be
        // able to mark a newer one dead — reader.Join can time out, and the stale thread's
        // finally block would otherwise tear down the connection that replaced it.
        private static int _epoch;

        /// <summary>
        ///     Loopback reconnects are cheap, so the ceiling is low. A sidecar restart should cost
        ///     a few seconds of buffering, not half a minute.
        /// </summary>
        private const int MaxBackoffMs = 5000;

        private static int _depth;
        private static long _sent, _dropped, _received, _connects, _writeErrors;

        /// <summary>Raised on the <b>Core thread</b>, one call per inbound line.</summary>
        public static event Action<string> InboundLine;

        /// <summary>
        ///     Raised on the <b>Core thread</b> after each successful connect. The sidecar may
        ///     restart independently of the shard, so anything it needs to know up front has to be
        ///     re-sent per connection, not once at ServerStarted.
        /// </summary>
        public static event Action Connected_Core;

        public static bool Connected { get { return _connected; } }
        public static int Depth { get { return Volatile.Read(ref _depth); } }
        public static long Sent { get { return Interlocked.Read(ref _sent); } }
        public static long Dropped { get { return Interlocked.Read(ref _dropped); } }
        public static long Received { get { return Interlocked.Read(ref _received); } }
        /// <summary>Total successful connections, so the first connect counts as 1.</summary>
        public static long Connects { get { return Interlocked.Read(ref _connects); } }
        public static long WriteErrors { get { return Interlocked.Read(ref _writeErrors); } }

        public static void Start()
        {
            if (_running)
                return;

            _running = true;

            _link = new Thread(LinkLoop)
            {
                Name = "Bridge Link",
                IsBackground = true
            };

            _link.Start();
        }

        public static void Stop()
        {
            if (!_running)
                return;

            _running = false;
            _wake.Set();

            var t = _link;

            if (t != null && !t.Join(TimeSpan.FromSeconds(2.0)))
                Console.WriteLine("[Bridge] link thread did not stop cleanly");

            _link = null;
            _connected = false;
        }

        /// <summary>
        ///     Core thread. Non-blocking. `line` must already be a complete JSON object with no
        ///     embedded newline; the newline is appended by the writer as the frame delimiter.
        /// </summary>
        public static void Emit(string line)
        {
            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
            // the cap but never grows without limit.
            while (Volatile.Read(ref _depth) >= BridgeConfig.QueueCap)
            {
                string discard;

                if (!_outbound.TryDequeue(out discard))
                    break;

                Interlocked.Decrement(ref _depth);
                Interlocked.Increment(ref _dropped);
            }

            _outbound.Enqueue(line);
            Interlocked.Increment(ref _depth);

            _wake.Set();
        }

        private static void LinkLoop()
        {
            int backoffMs = 500;

            while (_running)
            {
                TcpClient client = null;
                Thread reader = null;

                int epoch = Interlocked.Increment(ref _epoch);

                try
                {
                    client = new TcpClient();
                    client.NoDelay = true;
                    client.Connect(BridgeConfig.Host, BridgeConfig.Port);

                    var stream = client.GetStream();
                    stream.WriteTimeout = 5000; // a wedged peer must surface as an error, not a hang

                    _dead = false;
                    _connected = true;
                    backoffMs = 500;

                    Interlocked.Increment(ref _connects);
                    Console.WriteLine("[Bridge] connected to {0}:{1}", BridgeConfig.Host, BridgeConfig.Port);

                    // Building the greeting reads the world, so it must happen on the Core thread.
                    Timer.DelayCall(TimeSpan.Zero, () =>
                    {
                        try
                        {
                            var handler = Connected_Core;

                            if (handler != null)
                                handler();
                        }
                        catch (Exception ex)
                        {
                            Console.WriteLine("[Bridge] connect handler threw: {0}", ex);
                        }
                    });

                    var localStream = stream;
                    reader = new Thread(() => ReadLoop(localStream, epoch))
                    {
                        Name = "Bridge Reader",
                        IsBackground = true
                    };
                    reader.Start();

                    WriteLoop(stream);
                }
                catch (Exception ex)
                {
                    if (_connected)
                        Console.WriteLine("[Bridge] link error: {0}", ex.Message);
                }
                finally
                {
                    _connected = false;
                    _dead = true;

                    try { if (client != null) client.Close(); }
                    catch { }

                    if (reader != null)
                        reader.Join(TimeSpan.FromSeconds(1.0));
                }

                if (!_running)
                    break;

                // Nothing is listening yet, or the sidecar restarted. Both are normal.
                Thread.Sleep(backoffMs);
                backoffMs = Math.Min(backoffMs * 2, MaxBackoffMs);
            }

            _connected = false;
        }

        private static void WriteLoop(NetworkStream stream)
        {
            while (_running && !_dead)
            {
                string line;

                if (!_outbound.TryDequeue(out line))
                {
                    _wake.WaitOne(250);
                    continue;
                }

                Interlocked.Decrement(ref _depth);

                try
                {
                    var bytes = Encoding.UTF8.GetBytes(line + "\n");
                    stream.Write(bytes, 0, bytes.Length);
                    Interlocked.Increment(ref _sent);
                }
                catch (Exception)
                {
                    // The record is already off the queue. Count it and let the outer loop
                    // reconnect; re-queueing risks an unbounded retry storm against a dead peer.
                    Interlocked.Increment(ref _writeErrors);
                    _dead = true;
                    throw;
                }
            }
        }

        private static void ReadLoop(NetworkStream stream, int epoch)
        {
            var buffer = new byte[8192];
            var line = new StringBuilder(512);

            try
            {
                while (_running && !_dead)
                {
                    int read = stream.Read(buffer, 0, buffer.Length);

                    if (read <= 0)
                        break; // clean EOF: the sidecar closed. Normal.

                    for (int i = 0; i < read; i++)
                    {
                        char c = (char)buffer[i];

                        if (c == '\n')
                        {
                            Dispatch(line.ToString());
                            line.Clear();
                        }
                        else if (c != '\r')
                        {
                            line.Append(c);

                            if (line.Length > (1 << 20))
                            {
                                Console.WriteLine("[Bridge] inbound line too long, dropping");
                                line.Clear();
                            }
                        }
                    }
                }
            }
            catch (IOException)
            {
                // Expected when the peer vanishes mid-read.
            }
            catch (ObjectDisposedException)
            {
                // Expected when Stop() closes the socket under us.
            }
            catch (Exception ex)
            {
                Console.WriteLine("[Bridge] reader error: {0}", ex.Message);
            }
            finally
            {
                // Only tear down the connection this reader actually owned.
                if (Volatile.Read(ref _epoch) == epoch)
                {
                    _dead = true;
                    _wake.Set(); // let the writer notice and fall through to reconnect
                }
            }
        }

        /// <summary>
        ///     Reader thread. Marshals to the Core thread. Timer.DelayCall's scheduling path is
        ///     lock-protected and safe to call from any thread; the callback runs on the main loop.
        /// </summary>
        private static void Dispatch(string line)
        {
            if (line.Length == 0)
                return;

            Interlocked.Increment(ref _received);

            Timer.DelayCall(TimeSpan.Zero, () =>
            {
                try
                {
                    var handler = InboundLine;

                    if (handler != null)
                        handler(line);
                }
                catch (Exception ex)
                {
                    // A malformed command must never escape into a game code path.
                    Console.WriteLine("[Bridge] inbound handler threw: {0}", ex);
                }
            });
        }
    }
}