lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbWalVisibility.cs

using System.Diagnostics;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;

namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;

/// <summary>
///     The read-your-writes wait (R5, T0 Q4). QuestDB applies WAL commits asynchronously, so a row just inserted may
///     not be visible to the next query. After its INSERT(s) a writer calls <see cref="WaitAsync"/>, which captures
///     <c>sequencerTxn</c> from <c>wal_tables()</c> (every autocommitted INSERT is one sequencer transaction; a concurrent
///     writer can only raise the target) and polls <c>writerTxn</c> every ~10 ms until it reaches the target, the table
///     is suspended, or <see cref="QuestDbHistorianOptions.VisibilityTimeout"/> elapses.
/// </summary>
internal sealed class QuestDbWalVisibility
{
    private static readonly TimeSpan PollInterval = TimeSpan.FromMilliseconds(10);

    private readonly QuestDbConnection _conn;
    private readonly TimeSpan _timeout;
    private readonly ILogger _log;

    /// <summary>Creates the waiter.</summary>
    /// <param name="conn">Connection helper.</param>
    /// <param name="timeout">Maximum wait; <see cref="TimeSpan.Zero"/> (or less) disables the wait.</param>
    /// <param name="log">Optional logger.</param>
    public QuestDbWalVisibility(QuestDbConnection conn, TimeSpan timeout, ILogger? log = null)
    {
        _conn = conn;
        _timeout = timeout;
        _log = log ?? NullLogger.Instance;
    }

    /// <summary>True when the wait is disabled (<c>VisibilityTimeout</c> is 0).</summary>
    public bool Disabled => _timeout <= TimeSpan.Zero;

    /// <summary>True when the applied transaction has caught up with the target, or the table cannot progress.</summary>
    /// <param name="writerTxn">Last applied transaction.</param>
    /// <param name="targetTxn">The sequencer transaction captured after the INSERT.</param>
    /// <returns>Whether polling can stop because the rows are visible.</returns>
    internal static bool Reached(long writerTxn, long targetTxn) => writerTxn >= targetTxn;

    /// <summary>Waits until everything this caller inserted into <paramref name="table"/> is visible to queries.</summary>
    /// <param name="table">Full table name written to.</param>
    /// <param name="ct">Cancellation.</param>
    /// <returns>
    ///     True when visible (or the wait is disabled, or the table is not a WAL table); false when the table is suspended
    ///     or the timeout elapsed (a Warning is logged; the rows are still committed and become visible later).
    /// </returns>
    public async Task<bool> WaitAsync(string table, CancellationToken ct)
    {
        if (Disabled) return true;
        var sw = Stopwatch.StartNew();
        await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);

        long target;
        await using (var cmd = _conn.CreateCommand(conn, "SELECT sequencerTxn FROM wal_tables() WHERE name = $1"))
        {
            QuestDbConnection.BindText(cmd, table);
            var o = await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false);
            if (o is null or DBNull) return true; // not a WAL table: writes are immediately visible
            target = Convert.ToInt64(o);
        }

        while (true)
        {
            long writer;
            bool suspended;
            await using (var cmd = _conn.CreateCommand(conn, "SELECT writerTxn, suspended FROM wal_tables() WHERE name = $1"))
            {
                QuestDbConnection.BindText(cmd, table);
                await using var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
                if (!await r.ReadAsync(ct).ConfigureAwait(false)) return true;
                writer = r.GetInt64(0);
                suspended = r.GetBoolean(1);
            }
            if (Reached(writer, target)) return true;
            if (suspended)
            {
                _log.LogWarning("QuestDB table {Table} is suspended (writerTxn {Writer} < sequencerTxn {Target}); not waiting.",
                    table, writer, target);
                return false;
            }
            if (sw.Elapsed >= _timeout)
            {
                _log.LogWarning("QuestDB table {Table} did not become visible within {Timeout} ms (writerTxn {Writer} < sequencerTxn {Target}).",
                    table, _timeout.TotalMilliseconds, writer, target);
                return false;
            }
            await Task.Delay(PollInterval, ct).ConfigureAwait(false);
        }
    }
}