fix(historian): QuestDB modification trail gets an applied marker and a dedup key (lmxopcua#746)

QuestDB has no multi-table transaction, so a HistoryUpdate writes its trail row before the sample change. A failed or crashed change left an orphan trail row that modified reads served as a change that happened, and a trail INSERT re-sent after a client timeout duplicated (the table had no key). - sample_modified gains `pending` and DEDUP UPSERT KEYS(ts, tag, modified_id). The trail row is written pending; after the change lands the same row is re-upserted pending = false (the applied marker). - A transient trail INSERT failure is re-sent once with the same ids; a first attempt that did apply converges on the key. - ReadModifiedAsync serves a pending row only when it is the latest trail row at its time and the sample there was written at or after the modification (a crash between change and marker). The orphan of a change that never landed, typically followed by the client's retry, is dropped. ASOF JOIN over the window's samples of one tag; the resume filter sits outside the window function so it never cuts a time's partition. - The bootstrapper migrates an older trail (ADD COLUMN pending, DEDUP ENABLE) and waits for it to apply (bounded 10 s); old rows read pending = false and stay visible. Until applied, or with AutoCreateSchema=false, the old format is used. Rebased from the unpushed remed/lane-H2 branch (originally authored 2026-10-22 in session_01Ub9hGmYLQvn9XETzw7CMsT) onto current master. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
2026-10-22 17:20:19 -04:00
parent 9c3c9b0d61
commit 1a7a8f6f22
8 changed files with 462 additions and 80 deletions
@@ -192,25 +192,44 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
                var start = QuestDbConnection.Round(startUtc);
                var end = QuestDbConnection.Round(endUtc);
                var reverse = start > end;
-                string where;
-                if (start == end) where = "ts = $2 AND ts = $3";
-                else if (!reverse) where = "ts >= $2 AND ts < $3";
-                else where = "ts > $2 AND ts <= $3";
+                string window;
+                if (start == end) window = "ts = $2 AND ts = $3";
+                else if (!reverse) window = "ts >= $2 AND ts < $3";
+                else window = "ts > $2 AND ts <= $3";
+                var resume = "";
                var next = 4;
                if (resumeAfter is not null)
                {
-                    where += reverse
+                    resume = reverse
                        ? " AND (ts < $4 OR (ts = $4 AND modified_id < $5))"
                        : " AND (ts > $4 OR (ts = $4 AND modified_id > $5))";
                    next = 6;
                }
                var order = reverse ? "DESC" : "ASC";
                var (lo, hi) = start <= end ? (start, end) : (end, start);
+                const string Columns = "ts, kind, num_value, int_value, str_value, status, server_ts, modified_id, " +
+                                       "modification_ts, update_type, principal";
+
+                // The applied marker: a trail row still `pending` is served only when the change it describes evidently
+                // landed — it is the LATEST row at its time and the sample there (a tombstone included) was written at or
+                // after the modification: a crash between the change and its marker. Otherwise it is the orphan of a change
+                // that never landed (typically followed by the client's successful retry, which is then the latest row) and
+                // is dropped. The ASOF JOIN reads the tag's samples of the window only (indexed tag, inclusive bounds); the
+                // resume position filters OUTSIDE the window function so it never cuts a time's partition.
+                var sql = _bootstrap.SupportsTrailMarker
+                    ? $"SELECT {Columns} FROM (SELECT m.ts ts, m.kind kind, m.num_value num_value, m.int_value int_value, " +
+                      "m.str_value str_value, m.status status, m.server_ts server_ts, m.modified_id modified_id, " +
+                      "m.modification_ts modification_ts, m.update_type update_type, m.principal principal, m.pending pending, " +
+                      "s.ts sample_ts, s.server_ts sample_server_ts, max(m.modified_id) OVER (PARTITION BY m.ts) latest_id " +
+                      $"FROM {_conn.SampleModifiedTable} m ASOF JOIN (SELECT ts, tag, server_ts FROM {_conn.SampleTable} " +
+                      $"WHERE tag = $1 AND ts >= $2 AND ts <= $3) s ON (tag) WHERE m.tag = $1 AND {window.Replace("ts ", "m.ts ")}) " +
+                      "WHERE (NOT pending OR (sample_ts = ts AND sample_server_ts >= modification_ts AND modified_id = latest_id))" +
+                      $"{resume} ORDER BY ts {order}, modified_id {order}"
+                    : $"SELECT {Columns} FROM {_conn.SampleModifiedTable} " +
+                      $"WHERE tag = $1 AND {window}{resume} ORDER BY ts {order}, modified_id {order}";

                await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false);
-                await using var cmd = _conn.CreateCommand(conn,
-                    $"SELECT {SampleColumns}, modified_id, modification_ts, update_type, principal FROM {_conn.SampleModifiedTable} " +
-                    $"WHERE tag = $1 AND {where} ORDER BY ts {order}, modified_id {order}" + (bounded ? $" LIMIT ${next}" : ""));
+                await using var cmd = _conn.CreateCommand(conn, sql + (bounded ? $" LIMIT ${next}" : ""));
                QuestDbConnection.BindText(cmd, fullReference);
                QuestDbConnection.BindTimestamp(cmd, lo);
                QuestDbConnection.BindTimestamp(cmd, hi);
@@ -32,6 +32,13 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
 ///     no multi-table transaction, so the trail row is written <b>BEFORE</b> the change: a crash (or a failed sample INSERT)
 ///     between the two leaves an orphan trail row describing a change that never landed — never a change without its
 ///     trail row. This is the documented deviation from SQL Server, whose trail commits atomically with the change.</para>
+///     <para><b>Applied marker.</b> On a trail table with the marker format
+///     (<see cref="QuestDbSchemaBootstrapper.SupportsTrailMarker"/>) the trail row is written <c>pending</c> and, once the
+///     sample change has landed, re-upserted with <c>pending = false</c> (same <c>(ts, tag, modified_id)</c> — the table's
+///     dedup key). A modified read drops a pending row unless the sample at its time was written at or after the
+///     modification (the change landed but the marker did not — a crash in between), so an orphan is no longer served as
+///     a change that happened. The trail INSERT is re-sent once (same ids) after a transient failure: a first attempt that
+///     did apply converges through the dedup key instead of duplicating.</para>
 /// </summary>
 internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
 {
@@ -154,10 +161,11 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
            if (page.Count == 0) break;
            // Trail BEFORE the change (no multi-table transaction): a crash here leaves an orphan trail row, never a
            // tombstone without its trail row.
-            await WriteTrailAsync(tag, HistoryModificationType.Delete,
+            var trail = await WriteTrailAsync(tag, HistoryModificationType.Delete,
                priors.Select(r => (r.TimeUtc, (LiveRow?)r)).ToList(), context, ct).ConfigureAwait(false);
            await AfterTrailAsync(tag, page[0], page[^1], page.Count,
                () => WriteTombstonesAsync(tag, page, ct)).ConfigureAwait(false);
+            await MarkAppliedAsync(tag, trail, ct).ConfigureAwait(false);
            // The wait between pages is required: without it the next page could re-read rows just tombstoned.
            await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false);
            total += page.Count;
@@ -197,11 +205,12 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
        foreach (var chunk in toDelete.Chunk(BatchSize))
        {
            // Trail BEFORE the change (see the class remarks).
-            await WriteTrailAsync(tag, HistoryModificationType.Delete,
+            var trail = await WriteTrailAsync(tag, HistoryModificationType.Delete,
                chunk.Select(t => (live[t].TimeUtc, (LiveRow?)live[t])).ToList(), context, ct).ConfigureAwait(false);
            var chunkTimes = chunk.Select(t => new DateTime(t, DateTimeKind.Utc)).ToList();
            await AfterTrailAsync(tag, chunkTimes[0], chunkTimes[^1], chunkTimes.Count,
                () => WriteTombstonesAsync(tag, chunkTimes, ct)).ConfigureAwait(false);
+            await MarkAppliedAsync(tag, trail, ct).ConfigureAwait(false);
        }
        if (toDelete.Count > 0) await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false);
        deleted.UnionWith(toDelete);
@@ -233,10 +242,11 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
            trail.Add((stored, present ? prior : null));
        }
        // Trail BEFORE the change (see the class remarks).
-        await WriteTrailAsync(tag, ToModificationType(mode), trail, context, ct).ConfigureAwait(false);
+        var written = await WriteTrailAsync(tag, ToModificationType(mode), trail, context, ct).ConfigureAwait(false);
        foreach (var chunk in toWrite.Chunk(BatchSize))
            await AfterTrailAsync(tag, chunk[0].TimeUtc, chunk[^1].TimeUtc, chunk.Length,
                () => WriteRowsAsync(tag, chunk, ct)).ConfigureAwait(false);
+        await MarkAppliedAsync(tag, written, ct).ConfigureAwait(false);
        if (toWrite.Count > 0) await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false);
        return outcome;
    }
@@ -268,7 +278,7 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
    private const string LiveColumns = "ts, kind, num_value, int_value, str_value, status, server_ts";

    /// <summary>A live sample as it was before a modification.</summary>
-    private readonly record struct LiveRow(DateTime TimeUtc, HistorianValue Value, uint Status, DateTime? ServerTimeUtc);
+    internal readonly record struct LiveRow(DateTime TimeUtc, HistorianValue Value, uint Status, DateTime? ServerTimeUtc);

    private static LiveRow ReadLiveRow(NpgsqlDataReader rd)
    {
@@ -305,63 +315,38 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
        }
    }

+    /// <summary>One modification-trail row as written: its key, its prior value and the fields re-sent by the marker.</summary>
+    internal sealed record TrailRow(DateTime TimeUtc, LiveRow? Prior, long ModifiedId, HistoryModificationType Type,
+        DateTime ModificationUtc, string Principal, string? ApiKeyId);
+
+    /// <summary>Test seam: runs after a trail INSERT executed (once per attempt), before its outcome is decided.</summary>
+    internal Func<int, Task>? AfterTrailInsert { get; set; }
+
+    /// <summary>Test seam: runs before each sample write that follows a written trail batch (a throw = a failed change).</summary>
+    internal Func<Task>? BeforeSampleWrite { get; set; }
+
+    /// <summary>Test seam: when true the applied marker is never written (a crash after the change, before the marker).</summary>
+    internal bool SkipAppliedMarker { get; set; }
+
    /// <summary>
    ///     Appends one modification-trail row per entry (<c>prior = null</c> ⇒ an insert: kind Empty, NULL value and
-    ///     status), then waits for the trail's WAL visibility so a modified read right after the update sees it.
+    ///     status) — <c>pending</c> on a marker-format trail — then waits for the trail's WAL visibility so a modified read
+    ///     right after the update sees it. Returns the rows written, for <see cref="MarkAppliedAsync"/>.
    /// </summary>
-    private async Task WriteTrailAsync(string tag, HistoryModificationType type, IReadOnlyList<(DateTime TimeUtc, LiveRow? Prior)> entries,
-        HistoryUpdateContext context, CancellationToken ct)
+    private async Task<IReadOnlyList<TrailRow>> WriteTrailAsync(string tag, HistoryModificationType type,
+        IReadOnlyList<(DateTime TimeUtc, LiveRow? Prior)> entries, HistoryUpdateContext context, CancellationToken ct)
    {
-        if (entries.Count == 0) return;
+        if (entries.Count == 0) return [];
        var now = DateTime.UtcNow;
-        foreach (var chunk in entries.Chunk(BatchSize))
+        var rows = entries
+            .Select(e => new TrailRow(e.TimeUtc, e.Prior, NextModifiedId(), type, now, context.StoredPrincipal, context.StoredApiKeyId))
+            .ToList();
+        var marker = _tags.Bootstrap.SupportsTrailMarker;
+        foreach (var chunk in rows.Chunk(BatchSize))
        {
-            await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
-            await using var cmd = _conn.CreateCommand(conn, "");
-            var sb = new System.Text.StringBuilder(
-                $"INSERT INTO {_conn.SampleModifiedTable} (ts, tag, modified_id, modification_ts, update_type, principal, " +
-                "api_key_id, kind, num_value, int_value, str_value, status, server_ts) VALUES ");
-            for (var i = 0; i < chunk.Length; i++)
-            {
-                var (time, prior) = chunk[i];
-                QuestDbConnection.BindTimestamp(cmd, time, clampToEpoch: true);
-                QuestDbConnection.BindText(cmd, tag);
-                QuestDbConnection.BindLong(cmd, NextModifiedId());
-                QuestDbConnection.BindTimestamp(cmd, now);
-                QuestDbConnection.BindByte(cmd, (byte)type);
-                QuestDbConnection.BindText(cmd, context.StoredPrincipal);
-                QuestDbConnection.BindText(cmd, context.StoredApiKeyId);
-                if (prior is { } p)
-                {
-                    var c = SampleColumnMapper.ToColumns(p.Value);
-                    QuestDbConnection.BindByte(cmd, c.ValueKind);
-                    Typed(cmd, NpgsqlDbType.Double, c.NumValue);
-                    Typed(cmd, NpgsqlDbType.Bigint, c.IntValue);
-                    Typed(cmd, NpgsqlDbType.Varchar, c.StrValue);
-                    QuestDbConnection.BindLong(cmd, SampleColumnMapper.ToLongStatus(p.Status));
-                    Typed(cmd, NpgsqlDbType.TimestampTz, p.ServerTimeUtc);
-                }
-                else
-                {
-                    // No prior value: QuestDB BYTE has no NULL, so kind is Empty (0) and the NULL status marks "none".
-                    QuestDbConnection.BindByte(cmd, (byte)HistorianValueKind.Empty);
-                    Typed(cmd, NpgsqlDbType.Double, null);
-                    Typed(cmd, NpgsqlDbType.Bigint, null);
-                    Typed(cmd, NpgsqlDbType.Varchar, null);
-                    Typed(cmd, NpgsqlDbType.Bigint, null);
-                    Typed(cmd, NpgsqlDbType.TimestampTz, null);
-                }
-
-                var b = i * 13;
-                if (i > 0) sb.Append(',');
-                sb.Append('(');
-                for (var q = 1; q <= 13; q++) sb.Append(q > 1 ? "," : "").Append('$').Append(b + q);
-                sb.Append(')');
-            }
-            cmd.CommandText = sb.ToString();
            try
            {
-                await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
+                await InsertTrailAsync(tag, chunk, pending: marker, retryOnce: marker, ct).ConfigureAwait(false);
            }
            catch (Exception ex) when (ex is not OperationCanceledException)
            {
@@ -371,6 +356,104 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
            }
        }
        await _wal.WaitAsync(_conn.SampleModifiedTable, ct).ConfigureAwait(false);
+        return rows;
+    }
+
+    /// <summary>
+    ///     The applied marker: re-upserts the trail rows of a change that has landed with <c>pending = false</c> (same
+    ///     <c>(ts, tag, modified_id)</c> key). A failure here does not fail the update — the change landed, and a modified read
+    ///     still serves a pending row whose sample shows the change (logged as a Warning).
+    /// </summary>
+    private async Task MarkAppliedAsync(string tag, IReadOnlyList<TrailRow> rows, CancellationToken ct)
+    {
+        if (rows.Count == 0 || !_tags.Bootstrap.SupportsTrailMarker || SkipAppliedMarker) return;
+        try
+        {
+            foreach (var chunk in rows.Chunk(BatchSize))
+                await InsertTrailAsync(tag, chunk, pending: false, retryOnce: true, ct).ConfigureAwait(false);
+            await _wal.WaitAsync(_conn.SampleModifiedTable, ct).ConfigureAwait(false);
+        }
+        catch (Exception ex) when (ex is not OperationCanceledException)
+        {
+            _log.LogWarning(ex, "QuestDB applied marker for {Count} trail row(s) of tag {Tag} was not written; the change landed and " +
+                "modified reads still serve it.", rows.Count, tag);
+        }
+    }
+
+    private int _trailInserts;
+
+    /// <summary>Trail INSERT attempts so far (lifetime) — test/diagnostic.</summary>
+    internal int TrailInserts => Volatile.Read(ref _trailInserts);
+
+    /// <summary>One multi-row trail INSERT; re-sent once with the same values after a transient failure when allowed.</summary>
+    private async Task InsertTrailAsync(string tag, IReadOnlyList<TrailRow> chunk, bool pending, bool retryOnce, CancellationToken ct)
+    {
+        for (var attempt = 0; ; attempt++)
+        {
+            try
+            {
+                await InsertTrailOnceAsync(tag, chunk, pending, ct).ConfigureAwait(false);
+                Interlocked.Increment(ref _trailInserts);
+                if (AfterTrailInsert is { } hook) await hook(attempt).ConfigureAwait(false);
+                return;
+            }
+            catch (Exception ex) when (retryOnce && attempt == 0 && !ct.IsCancellationRequested && QuestDbConnection.IsTransient(ex))
+            {
+                // The first attempt may have applied (a client timeout after the server took it): the same ids re-sent
+                // converge through DEDUP UPSERT KEYS(ts, tag, modified_id) instead of duplicating.
+                _log.LogDebug("QuestDB modification-trail INSERT failed transiently ({Exception}); re-sending it once.", ex.GetType().Name);
+            }
+        }
+    }
+
+    private async Task InsertTrailOnceAsync(string tag, IReadOnlyList<TrailRow> chunk, bool pending, CancellationToken ct)
+    {
+        var columns = _tags.Bootstrap.SupportsTrailMarker ? 14 : 13;
+        await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
+        await using var cmd = _conn.CreateCommand(conn, "");
+        var sb = new System.Text.StringBuilder(
+            $"INSERT INTO {_conn.SampleModifiedTable} (ts, tag, modified_id, modification_ts, update_type, principal, " +
+            "api_key_id, kind, num_value, int_value, str_value, status, server_ts" + (columns == 14 ? ", pending" : "") + ") VALUES ");
+        for (var i = 0; i < chunk.Count; i++)
+        {
+            var row = chunk[i];
+            QuestDbConnection.BindTimestamp(cmd, row.TimeUtc, clampToEpoch: true);
+            QuestDbConnection.BindText(cmd, tag);
+            QuestDbConnection.BindLong(cmd, row.ModifiedId);
+            QuestDbConnection.BindTimestamp(cmd, row.ModificationUtc);
+            QuestDbConnection.BindByte(cmd, (byte)row.Type);
+            QuestDbConnection.BindText(cmd, row.Principal);
+            QuestDbConnection.BindText(cmd, row.ApiKeyId);
+            if (row.Prior is { } p)
+            {
+                var c = SampleColumnMapper.ToColumns(p.Value);
+                QuestDbConnection.BindByte(cmd, c.ValueKind);
+                Typed(cmd, NpgsqlDbType.Double, c.NumValue);
+                Typed(cmd, NpgsqlDbType.Bigint, c.IntValue);
+                Typed(cmd, NpgsqlDbType.Varchar, c.StrValue);
+                QuestDbConnection.BindLong(cmd, SampleColumnMapper.ToLongStatus(p.Status));
+                Typed(cmd, NpgsqlDbType.TimestampTz, p.ServerTimeUtc);
+            }
+            else
+            {
+                // No prior value: QuestDB BYTE has no NULL, so kind is Empty (0) and the NULL status marks "none".
+                QuestDbConnection.BindByte(cmd, (byte)HistorianValueKind.Empty);
+                Typed(cmd, NpgsqlDbType.Double, null);
+                Typed(cmd, NpgsqlDbType.Bigint, null);
+                Typed(cmd, NpgsqlDbType.Varchar, null);
+                Typed(cmd, NpgsqlDbType.Bigint, null);
+                Typed(cmd, NpgsqlDbType.TimestampTz, null);
+            }
+            if (columns == 14) Typed(cmd, NpgsqlDbType.Boolean, pending);
+
+            var b = i * columns;
+            if (i > 0) sb.Append(',');
+            sb.Append('(');
+            for (var q = 1; q <= columns; q++) sb.Append(q > 1 ? "," : "").Append('$').Append(b + q);
+            sb.Append(')');
+        }
+        cmd.CommandText = sb.ToString();
+        await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
    }

    /// <summary>
@@ -382,6 +465,7 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
    {
        try
        {
+            if (BeforeSampleWrite is { } hook) await hook().ConfigureAwait(false);
            await write().ConfigureAwait(false);
        }
        catch (Exception ex)
@@ -73,11 +73,13 @@ internal sealed class QuestDbSchemaBootstrapper
                    await CreateTablesAsync(conn, ct).ConfigureAwait(false);
                    await ReconcileTtlAsync(conn, ct).ConfigureAwait(false);
                    await EnsureIndexesAsync(conn, apply: true, ct).ConfigureAwait(false);
+                    SupportsTrailMarker = await EnsureTrailMarkerAsync(conn, apply: true, ct).ConfigureAwait(false);
                }
                else
                {
                    await VerifyAsync(conn, ct).ConfigureAwait(false);
                    await EnsureIndexesAsync(conn, apply: false, ct).ConfigureAwait(false);
+                    SupportsTrailMarker = await EnsureTrailMarkerAsync(conn, apply: false, ct).ConfigureAwait(false);
                }
                _ready = true;
                _failure.Clear();
@@ -128,6 +130,93 @@ internal sealed class QuestDbSchemaBootstrapper
        }
    }

+    private volatile bool _supportsTrailMarker;
+
+    /// <summary>
+    ///     Whether the modification trail has its <c>pending</c> column and <c>DEDUP UPSERT KEYS(ts, tag, modified_id)</c>
+    ///     (true for every table created since; a pre-marker table once this boot's migration has applied). False ⇒ the
+    ///     updater writes the trail the old way (no marker, no retry) and modified reads do not reconcile — set by
+    ///     <see cref="EnsureAsync"/>.
+    /// </summary>
+    internal bool SupportsTrailMarker
+    {
+        get => _supportsTrailMarker;
+        private set => _supportsTrailMarker = value;
+    }
+
+    /// <summary>How long a boot waits for its own trail migration (two WAL ALTERs) to apply before using the old format.</summary>
+    internal static readonly TimeSpan TrailMigrationWait = TimeSpan.FromSeconds(10);
+
+    /// <summary>
+    ///     Gives a pre-marker trail table its <c>pending</c> column and dedup key (<paramref name="apply"/>), then reports
+    ///     whether both are in effect. The ALTERs are WAL transactions, and an INSERT naming a column the table does not have
+    ///     yet would fail, so — unlike the TTL / index ALTERs — this one IS awaited (bounded by
+    ///     <see cref="TrailMigrationWait"/>); until it applies the old format is used. <paramref name="apply"/> false only
+    ///     logs the statements to run. A refused ALTER is a Warning, never a bootstrap failure.
+    /// </summary>
+    private async Task<bool> EnsureTrailMarkerAsync(NpgsqlConnection conn, bool apply, CancellationToken ct)
+    {
+        var table = _conn.SampleModifiedTable;
+        var (hasColumn, hasDedup) = await TrailStateAsync(conn, table, ct).ConfigureAwait(false);
+        if (hasColumn && hasDedup) return true;
+        var statements = new List<string>();
+        if (!hasColumn) statements.Add(QuestDbTables.AddTrailPendingColumn(table));
+        if (!hasDedup) statements.Add(QuestDbTables.EnableTrailDedup(table));
+        if (!apply)
+        {
+            _log.LogWarning("QuestDB modification trail {Table} predates its applied marker, so modified reads cannot drop the " +
+                "orphan of a HistoryUpdate that failed after its trail row, and a retried trail INSERT can duplicate. " +
+                "AutoCreateSchema is false; run: {Statements}", table, string.Join("; ", statements));
+            return false;
+        }
+        try
+        {
+            foreach (var sql in statements)
+            {
+                await using var alter = _conn.CreateCommand(conn, sql);
+                await alter.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
+            }
+        }
+        catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested)
+        {
+            _log.LogWarning("Could not migrate QuestDB modification trail {Table} ({Exception}: {Message}); the trail keeps its " +
+                "old format. Run: {Statements}", table, ex.GetType().Name, ex.Message, string.Join("; ", statements));
+            return false;
+        }
+        var deadline = DateTime.UtcNow + TrailMigrationWait;
+        while (true)
+        {
+            (hasColumn, hasDedup) = await TrailStateAsync(conn, table, ct).ConfigureAwait(false);
+            if (hasColumn && hasDedup)
+            {
+                _log.LogInformation("Migrated QuestDB modification trail {Table} to the applied-marker format.", table);
+                return true;
+            }
+            if (DateTime.UtcNow >= deadline)
+            {
+                _log.LogWarning("QuestDB modification trail {Table} migration did not apply within {Seconds} s; the old trail " +
+                    "format is used until the next boot.", table, TrailMigrationWait.TotalSeconds);
+                return false;
+            }
+            await Task.Delay(50, ct).ConfigureAwait(false);
+        }
+    }
+
+    private async Task<(bool Column, bool Dedup)> TrailStateAsync(NpgsqlConnection conn, string table, CancellationToken ct)
+    {
+        bool column;
+        await using (var cmd = _conn.CreateCommand(conn, QuestDbTables.ColumnCountQuery(table, "pending")))
+            column = Convert.ToInt64(await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false), CultureInfo.InvariantCulture) > 0;
+        bool dedup;
+        await using (var cmd = _conn.CreateCommand(conn, "SELECT dedup FROM tables() WHERE table_name = $1"))
+        {
+            QuestDbConnection.BindText(cmd, table);
+            var o = await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false);
+            dedup = o is not null and not DBNull && Convert.ToBoolean(o, CultureInfo.InvariantCulture);
+        }
+        return (column, dedup);
+    }
+
    /// <summary>Ensures every <see cref="QuestDbTables.SymbolIndexes"/> column is indexed (see <see cref="EnsureIndexAsync"/>).</summary>
    private async Task EnsureIndexesAsync(NpgsqlConnection conn, bool apply, CancellationToken ct)
    {
@@ -91,8 +91,13 @@ internal static class QuestDbTables
    ///     <c>CREATE TABLE IF NOT EXISTS</c> for the Part 11 modification trail: one row per sample a HistoryUpdate
    ///     inserted, replaced or deleted, carrying the PRIOR value (NULL kind/value/status for an insert). <c>ts</c> is the
    ///     sample's source time (so the TTL ages it out with the samples); <c>modified_id</c> orders the trail within one
-    ///     source time. Append-only: no DEDUP, never updated or deleted. <c>tag</c> carries a symbol index (every modified
-    ///     read filters on one tag; see <see cref="SymbolIndexes"/> for the migration of an older table).
+    ///     source time. Never updated or deleted; <c>tag</c> carries a symbol index (every modified read filters on one tag;
+    ///     see <see cref="SymbolIndexes"/> for the migration of an older table).
+    ///     <para><c>DEDUP UPSERT KEYS(ts, tag, modified_id)</c> + <c>pending</c>: a trail row is written <c>pending = true</c>
+    ///     BEFORE its sample change and re-upserted <c>pending = false</c> (the applied marker) after it, so a modified read
+    ///     can drop the orphan of a change that never landed; and a trail INSERT re-sent after a client timeout (same ids)
+    ///     converges instead of duplicating. A table from before both gains them through <see cref="AddTrailPendingColumn"/>
+    ///     / <see cref="EnableTrailDedup"/>.</para>
    /// </summary>
    /// <param name="prefix">The configured table-name prefix.</param>
    /// <param name="retentionDays">Retention in days, or null for no TTL.</param>
@@ -100,8 +105,25 @@ internal static class QuestDbTables
    public static string CreateSampleModified(string prefix, int? retentionDays) =>
        $"CREATE TABLE IF NOT EXISTS {SampleModified(prefix)} (ts TIMESTAMP, tag SYMBOL INDEX, modified_id LONG, " +
        "modification_ts TIMESTAMP, update_type BYTE, principal VARCHAR, api_key_id VARCHAR, kind BYTE, num_value DOUBLE, " +
-        "int_value LONG, str_value VARCHAR, status LONG, server_ts TIMESTAMP) " +
-        $"TIMESTAMP(ts) PARTITION BY DAY{Ttl(retentionDays)} WAL";
+        "int_value LONG, str_value VARCHAR, status LONG, server_ts TIMESTAMP, pending BOOLEAN) " +
+        $"TIMESTAMP(ts) PARTITION BY DAY{Ttl(retentionDays)} WAL DEDUP UPSERT KEYS(ts, tag, modified_id)";
+
+    /// <summary>The migration of a pre-marker trail table: <c>ALTER TABLE t ADD COLUMN pending BOOLEAN</c> (old rows read false).</summary>
+    /// <param name="table">The modification-trail table name.</param>
+    /// <returns>The ALTER statement.</returns>
+    public static string AddTrailPendingColumn(string table) => $"ALTER TABLE {table} ADD COLUMN pending BOOLEAN";
+
+    /// <summary>The migration of a pre-marker trail table: <c>ALTER TABLE t DEDUP ENABLE UPSERT KEYS(ts, tag, modified_id)</c>.</summary>
+    /// <param name="table">The modification-trail table name.</param>
+    /// <returns>The ALTER statement.</returns>
+    public static string EnableTrailDedup(string table) => $"ALTER TABLE {table} DEDUP ENABLE UPSERT KEYS(ts, tag, modified_id)";
+
+    /// <summary>Whether <paramref name="table"/> has <paramref name="column"/>: one <c>count()</c> row.</summary>
+    /// <param name="table">The table name (prefix-validated, inlined).</param>
+    /// <param name="column">The column (a constant).</param>
+    /// <returns>The query text.</returns>
+    public static string ColumnCountQuery(string table, string column) =>
+        $"SELECT count() FROM table_columns('{table}') WHERE \"column\" = '{column}'";

    /// <summary>The four CREATE statements in creation order (parallel to <see cref="All"/>).</summary>
    /// <param name="prefix">The configured table-name prefix.</param>
@@ -33,6 +33,9 @@ internal sealed class QuestDbTagCatalog
        _clock = clock ?? TimeProvider.System;
    }

+    /// <summary>The schema bootstrapper this catalog runs (its <see cref="QuestDbSchemaBootstrapper.SupportsTrailMarker"/>).</summary>
+    internal QuestDbSchemaBootstrapper Bootstrap => _bootstrap;
+
    /// <summary>Read-only lookup; never creates a row.</summary>
    /// <param name="name">Historian tag name.</param>
    /// <param name="ct">Cancellation.</param>