Whisper 0.1.0: private messages between players
/pm (/w) with names, #userid, SteamIDs, @me, @aim, @random, @randomt and @randomct, /r to reply, and /block, /blocklist, /unblock. Lines use CS2-Tags' rank prefixes and colours: >[[M] Kalus -> me] message. Gagged players can't send; blocks stop everyone but staff. Messages are kept in CS2-SimpleAdmin's database while History is on (css_whisper_history).
This commit is contained in:
commit
c414e8a770
12 changed files with 1547 additions and 0 deletions
287
Writer.cs
Normal file
287
Writer.cs
Normal file
|
|
@ -0,0 +1,287 @@
|
|||
using System.Text.Json;
|
||||
using System.Threading.Channels;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using MySqlConnector;
|
||||
|
||||
namespace Whisper;
|
||||
|
||||
// Runs every database write on one background task, in the order the game thread queued them. When
|
||||
// the database can't be reached, writes go to a spool file (one JSON op per line) instead, and every
|
||||
// later write follows them there until the spool has been sent, so the order never changes. Migrations
|
||||
// run first, retried until they work; until then everything is spooled.
|
||||
public sealed class Writer
|
||||
{
|
||||
private static readonly TimeSpan RetryInterval = TimeSpan.FromSeconds(15);
|
||||
|
||||
private readonly Database _db;
|
||||
private readonly string _spoolPath;
|
||||
private readonly int _spoolMax;
|
||||
private readonly ILogger _logger;
|
||||
private readonly Channel<Op> _queue = Channel.CreateUnbounded<Op>(new UnboundedChannelOptions { SingleReader = true });
|
||||
private readonly CancellationTokenSource _stop = new();
|
||||
private Task? _loop;
|
||||
|
||||
private bool _migrated;
|
||||
private bool _failing;
|
||||
private bool _spoolFullWarned;
|
||||
// While failing, migrations and the spool are only retried from this time on (UTC).
|
||||
private DateTime _retryAt;
|
||||
private volatile int _spooled;
|
||||
private volatile string? _lastError;
|
||||
private DateTime? _lastOkAt;
|
||||
|
||||
public Writer(Database db, string spoolPath, int spoolMax, ILogger logger)
|
||||
{
|
||||
_db = db;
|
||||
_spoolPath = spoolPath;
|
||||
_spoolMax = spoolMax;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
// Kept here: a single-reader unbounded channel can't count itself (Reader.Count throws).
|
||||
private int _queued;
|
||||
public int Queued => Volatile.Read(ref _queued);
|
||||
public int Spooled => _spooled;
|
||||
public bool Migrated => Volatile.Read(ref _migrated);
|
||||
public bool Failing => Volatile.Read(ref _failing);
|
||||
public string? LastError => _lastError;
|
||||
public DateTime? LastOkAt { get { lock (_stop) return _lastOkAt; } }
|
||||
|
||||
public void Start() => _loop = Task.Run(RunAsync);
|
||||
|
||||
// From the game thread. Never blocks.
|
||||
public void Enqueue(Op op)
|
||||
{
|
||||
if (_queue.Writer.TryWrite(op))
|
||||
Interlocked.Increment(ref _queued);
|
||||
}
|
||||
|
||||
// On unload: sends what's queued, waiting up to `timeout`, then spools anything left.
|
||||
public void Drain(TimeSpan timeout)
|
||||
{
|
||||
_queue.Writer.TryComplete();
|
||||
if (_loop == null)
|
||||
return;
|
||||
if (!_loop.Wait(timeout))
|
||||
{
|
||||
_stop.Cancel();
|
||||
if (!_loop.Wait(TimeSpan.FromSeconds(2)))
|
||||
{
|
||||
_logger.LogWarning("Database writer didn't stop; {Count} queued writes are lost", Queued);
|
||||
return;
|
||||
}
|
||||
}
|
||||
while (_queue.Reader.TryRead(out var op))
|
||||
{
|
||||
Interlocked.Decrement(ref _queued);
|
||||
Spool(op);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task RunAsync()
|
||||
{
|
||||
_spooled = CountSpool();
|
||||
if (_spooled > 0)
|
||||
_logger.LogInformation("{Count} writes waiting in the spool from before", _spooled);
|
||||
|
||||
var token = _stop.Token;
|
||||
while (!token.IsCancellationRequested)
|
||||
{
|
||||
if (!_failing || DateTime.UtcNow >= _retryAt)
|
||||
{
|
||||
if (!_migrated)
|
||||
await TryMigrateAsync(token);
|
||||
if (_migrated && _spooled > 0)
|
||||
await FlushSpoolAsync(token);
|
||||
}
|
||||
|
||||
while (_queue.Reader.TryRead(out var op))
|
||||
{
|
||||
Interlocked.Decrement(ref _queued);
|
||||
await HandleAsync(op, token);
|
||||
}
|
||||
|
||||
bool more;
|
||||
try
|
||||
{
|
||||
using var wait = CancellationTokenSource.CreateLinkedTokenSource(token);
|
||||
wait.CancelAfter(RetryInterval);
|
||||
more = await _queue.Reader.WaitToReadAsync(wait.Token);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
if (!more)
|
||||
{
|
||||
// Unloading: everything's been read. One last try at the spool.
|
||||
if (_migrated && _spooled > 0)
|
||||
await FlushSpoolAsync(token);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task TryMigrateAsync(CancellationToken token)
|
||||
{
|
||||
try
|
||||
{
|
||||
foreach (var version in await _db.MigrateAsync(token))
|
||||
_logger.LogInformation("Migration \"{Version}\" applied", version);
|
||||
Volatile.Write(ref _migrated, true);
|
||||
Succeeded();
|
||||
}
|
||||
catch (Exception ex) when (ex is not OperationCanceledException)
|
||||
{
|
||||
Failed(ex, "Can't reach the database to set up its tables; writes are spooled until it works");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleAsync(Op op, CancellationToken token)
|
||||
{
|
||||
if (!_migrated || _spooled > 0)
|
||||
{
|
||||
Spool(op);
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
await _db.ApplyAsync(op, token);
|
||||
Succeeded();
|
||||
}
|
||||
catch (Exception ex) when (Transient(ex))
|
||||
{
|
||||
Failed(ex, "Database write failed; spooling writes until it's back");
|
||||
Spool(op);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_lastError = ex.Message;
|
||||
_logger.LogError(ex, "Dropped a {Op} write the database refused", op.GetType().Name);
|
||||
}
|
||||
}
|
||||
|
||||
// Sends the spool in order. Stops at the first write that fails for a reason that may pass, and
|
||||
// keeps it and the rest.
|
||||
private async Task FlushSpoolAsync(CancellationToken token)
|
||||
{
|
||||
string[] lines;
|
||||
try
|
||||
{
|
||||
lines = File.ReadAllLines(_spoolPath);
|
||||
}
|
||||
catch (FileNotFoundException)
|
||||
{
|
||||
_spooled = 0;
|
||||
return;
|
||||
}
|
||||
|
||||
var sent = 0;
|
||||
for (; sent < lines.Length; sent++)
|
||||
{
|
||||
Op? op;
|
||||
try
|
||||
{
|
||||
op = JsonSerializer.Deserialize<Op>(lines[sent]);
|
||||
}
|
||||
catch (JsonException ex)
|
||||
{
|
||||
_logger.LogError(ex, "Skipped an unreadable spool line");
|
||||
continue;
|
||||
}
|
||||
if (op == null)
|
||||
continue;
|
||||
|
||||
try
|
||||
{
|
||||
await _db.ApplyAsync(op, token);
|
||||
Succeeded();
|
||||
}
|
||||
catch (Exception ex) when (Transient(ex))
|
||||
{
|
||||
Failed(ex, "Database write failed while sending the spool; trying again later");
|
||||
break;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_lastError = ex.Message;
|
||||
_logger.LogError(ex, "Dropped a spooled {Op} write the database refused", op.GetType().Name);
|
||||
}
|
||||
}
|
||||
|
||||
if (sent >= lines.Length)
|
||||
{
|
||||
File.Delete(_spoolPath);
|
||||
_spooled = 0;
|
||||
_logger.LogInformation("Sent {Count} spooled writes", lines.Length);
|
||||
return;
|
||||
}
|
||||
var temp = _spoolPath + ".tmp";
|
||||
File.WriteAllLines(temp, lines[sent..]);
|
||||
File.Move(temp, _spoolPath, overwrite: true);
|
||||
_spooled = lines.Length - sent;
|
||||
}
|
||||
|
||||
private void Spool(Op op)
|
||||
{
|
||||
if (_spooled >= _spoolMax)
|
||||
{
|
||||
if (!_spoolFullWarned)
|
||||
_logger.LogError("The spool is full ({Max} writes); new writes are dropped until the database is back", _spoolMax);
|
||||
_spoolFullWarned = true;
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
File.AppendAllText(_spoolPath, JsonSerializer.Serialize(op) + "\n");
|
||||
_spooled++;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Couldn't write to the spool {Path}; a {Op} write is lost", _spoolPath, op.GetType().Name);
|
||||
}
|
||||
}
|
||||
|
||||
private int CountSpool()
|
||||
{
|
||||
try
|
||||
{
|
||||
return File.ReadLines(_spoolPath).Count(l => l.Length > 0);
|
||||
}
|
||||
catch (FileNotFoundException)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
// Worth retrying: the database is unreachable, overloaded, or refusing our login (the password may
|
||||
// be fixed). Anything else (bad data, a missing table) would fail again forever.
|
||||
private static bool Transient(Exception ex) => ex switch
|
||||
{
|
||||
OperationCanceledException => true,
|
||||
MySqlException m => m.IsTransient || m.ErrorCode is MySqlErrorCode.UnableToConnectToHost
|
||||
or MySqlErrorCode.AccessDenied or MySqlErrorCode.DatabaseAccessDenied,
|
||||
TimeoutException or IOException or System.Net.Sockets.SocketException => true,
|
||||
_ => ex.InnerException != null && Transient(ex.InnerException),
|
||||
};
|
||||
|
||||
private void Succeeded()
|
||||
{
|
||||
lock (_stop)
|
||||
_lastOkAt = DateTime.UtcNow;
|
||||
_spoolFullWarned = false;
|
||||
if (Volatile.Read(ref _failing))
|
||||
_logger.LogInformation("Database is reachable again");
|
||||
Volatile.Write(ref _failing, false);
|
||||
}
|
||||
|
||||
// Logs once per outage, not per write.
|
||||
private void Failed(Exception ex, string message)
|
||||
{
|
||||
_lastError = ex.Message;
|
||||
_retryAt = DateTime.UtcNow + RetryInterval;
|
||||
if (!Volatile.Read(ref _failing))
|
||||
_logger.LogError("{Message}: {Error}", message, ex.Message);
|
||||
Volatile.Write(ref _failing, true);
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue