Phase 1: loopback transport to the sidecar
BridgeLink owns a TcpClient to 127.0.0.1 and nothing else touches it. Emit() is
called from the Core thread; it enqueues onto a bounded drop-oldest queue and
returns. A link thread drains the queue and reconnects with backoff; a reader
thread parses inbound lines and marshals each to the Core thread via
Timer.DelayCall. An absent, slow, or wedged sidecar therefore cannot stall the
shard, which is the property the rest of the bridge depends on.
Outbound JSON is written by hand into a StringBuilder because it runs on the
Core thread for every event and the measured budget assumes that cost. Inbound
uses JavaScriptSerializer: commands arrive at human rates, so correctness beats
speed, and parsing happens off the Core thread anyway. That needs a
System.Web.Extensions reference.
server.hello is emitted per connection rather than once at ServerStarted. A
sidecar that restarts independently would otherwise never learn which shard it
is attached to. It carries a bootId, stable across reconnects and fresh on every
shard restart, so the sidecar can tell "I reconnected" from "the shard
restarted" and keep or discard its cache accordingly.
Two defects found by testing and fixed before commit:
- Backoff ceiling was 30s, so a sidecar restart cost up to half a minute of
buffering on a loopback socket. Now 5s.
- A stale reader could kill a fresh connection: reader.Join(1s) can time out,
and the old thread's finally block then set the shared _dead flag, possibly
tearing down the connection that had replaced it. Connections now carry an
epoch and a reader only marks dead the one it owned.
Acceptance evidence recorded in docs/PLAN.md §11: boots with no sidecar, buffers
through the outage and drains on connect, round-trips ping/pong on the Core
thread, survives unknown kinds and malformed JSON, and reconnects unattended.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
331
overlay/Scripts/Custom/Bridge/BridgeLink.cs
Normal file
331
overlay/Scripts/Custom/Bridge/BridgeLink.cs
Normal file
@@ -0,0 +1,331 @@
|
||||
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 (!_running || line == null)
|
||||
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);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user