Whisper/Writer.cs

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