diff --git a/Darling/Darling.Tests/AggregateJobQuiesceTests.cs b/Darling/Darling.Tests/AggregateJobQuiesceTests.cs new file mode 100644 index 0000000000..357f55971e --- /dev/null +++ b/Darling/Darling.Tests/AggregateJobQuiesceTests.cs @@ -0,0 +1,852 @@ +/* + * Copyright (c) 2026 Erik Darling, Darling Data LLC + * + * This file is part of the SQL Server Performance Monitor. + * + * Licensed under the MIT License. See LICENSE file in the project root for full license information. + */ + +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Runtime.CompilerServices; +using System.Text.RegularExpressions; +using System.Threading; +using System.Threading.Tasks; +using Darling.Tests; +using Microsoft.Extensions.Logging; +using Npgsql; +using PerformanceMonitor.Darling.Storage; +using Xunit; + +namespace PerformanceMonitor.Darling.Tests; + +/// +/// #5551: the startup sweeps stop a continuous aggregate's jobs, and wait out a worker already running, before they +/// drop it. A refresh running during DROP MATERIALIZED VIEW ... CASCADE gave XX000: tuple concurrently +/// deleted (#5416) and is the inferred cause of a PostgreSQL backend crash in CI (#5549). These pins hold the SQL +/// text and the call order in the source; the live classes below prove the behaviour. +/// +[Trait("Stage", "Guard")] +public sealed class AggregateJobQuiescePinTests +{ + [Fact] + public void JobLookup_JoinsEitherJobIdentity_BindsTheViewName_AndFollowsDependentAggregates() + { + var sql = TimescaleSupport.ContinuousAggregateJobsSql; + + Assert.Contains("ca.view_schema = 'collect'", sql, StringComparison.Ordinal); + Assert.Contains("ca.view_name = $1::text", sql, StringComparison.Ordinal); + /* The view name is a parameter, never spliced into the text. */ + Assert.DoesNotContain("'{", sql, StringComparison.Ordinal); + /* Both identities a job can report (the user view, as timescaledb_information.jobs resolves it, and the + materialization hypertable) - measured, see ContinuousAggregateRefreshStateSql. */ + Assert.Contains("a.view_schema = j.hypertable_schema AND a.view_name = j.hypertable_name", sql, StringComparison.Ordinal); + Assert.Contains("a.mat_schema = j.hypertable_schema AND a.mat_name = j.hypertable_name", sql, StringComparison.Ordinal); + /* Every job TYPE of the aggregate (refresh, compression, retention): no proc_name filter. */ + Assert.DoesNotContain("proc_name", sql, StringComparison.Ordinal); + /* A hierarchical daily tier is dropped by the CASCADE too, so its jobs are found through the parent's + materialization hypertable, which is what its own hypertable_name reports. */ + Assert.Contains("WITH RECURSIVE", sql, StringComparison.Ordinal); + Assert.Contains("child.hypertable_name = parent.mat_name", sql, StringComparison.Ordinal); + Assert.Contains("j.scheduled", sql, StringComparison.Ordinal); + } + + [Fact] + public void Unschedule_NamesOnlyScheduled_AndReschedule_IsItsInverse_ByIntegerId() + { + Assert.Equal( + "SELECT alter_job(j.job_id, scheduled => false) FROM timescaledb_information.jobs AS j WHERE j.job_id = ANY($1::integer[])", + TimescaleSupport.UnscheduleJobsSql); + Assert.Equal( + "SELECT alter_job(j.job_id, scheduled => true) FROM timescaledb_information.jobs AS j WHERE j.job_id = ANY($1::integer[])", + TimescaleSupport.RescheduleJobsSql); + } + + [Fact] + public void WorkerCount_MatchesTheJobIdSuffix_InThisDatabaseOnly() + { + var sql = TimescaleSupport.AggregateJobWorkerCountSql; + + Assert.Contains("pg_stat_activity", sql, StringComparison.Ordinal); + Assert.Contains("a.datname = current_database()", sql, StringComparison.Ordinal); + Assert.Contains("a.backend_type LIKE '% [' || id::text || ']'", sql, StringComparison.Ordinal); + Assert.Contains("unnest($1::integer[])", sql, StringComparison.Ordinal); + } + + [Fact] + public void InformationViewProbe_GuardsAStoreWithoutTimescaleDb() + { + Assert.Contains("to_regclass('timescaledb_information.continuous_aggregates')", TimescaleSupport.ContinuousAggregateJobsProbeSql, StringComparison.Ordinal); + Assert.Contains("to_regclass('timescaledb_information.jobs')", TimescaleSupport.ContinuousAggregateJobsProbeSql, StringComparison.Ordinal); + } + + [Fact] + public void ProductionCap_IsThirtySeconds_AndFitsEverySweepRelationInsideTheHourlyBudget() + { + var options = TimescaleSupport.AggregateJobQuiesceOptions.Default; + + Assert.Equal(TimeSpan.FromSeconds(30), options.Cap); + Assert.Equal(TimeSpan.FromMilliseconds(250), options.PollInterval); + + /* Eight relations at most (two superseded, two retired, four reshaped), all stuck: 8 x 30 s = 240 s, inside + the hourly convergence pass's five-minute budget. */ + var relations = TimescaleSupport.SupersededBaselineRelations.Length + TimescaleSupport.RetiredBaselineRelations.Length + 4; + Assert.True(relations * options.Cap <= TimeSpan.FromMinutes(5), + $"{relations} relations x {options.Cap} must stay inside the hourly pass's budget"); + } + + /// + /// The census: every product statement that drops a continuous aggregate runs after + /// QuiesceContinuousAggregateJobsAsync in the same loop body, and a NEW drop site fails this test until + /// it is added here and given the same treatment. Comments are skipped; SQL text and statements are counted. + /// + [Fact] + public void EveryProductDropOfAContinuousAggregate_IsPrecededByTheQuiesce() + { + var root = RepoRoot(); + var productFiles = new[] { "PerformanceMonitor.Darling.Storage", "PerformanceMonitor.Darling.Service" } + .SelectMany(dir => Directory.EnumerateFiles(Path.Combine(root, "Darling", dir), "*.cs", SearchOption.AllDirectories)) + .Where(f => !f.Contains($"{Path.DirectorySeparatorChar}obj{Path.DirectorySeparatorChar}", StringComparison.Ordinal) + && !f.Contains($"{Path.DirectorySeparatorChar}bin{Path.DirectorySeparatorChar}", StringComparison.Ordinal)) + .ToArray(); + + var sites = new List<(string File, int Line)>(); + foreach (var file in productFiles) + { + var lines = File.ReadAllLines(file); + for (var i = 0; i < lines.Length; i++) + { + var trimmed = lines[i].TrimStart(); + /* A whole-line comment (//, ///, a block comment's first or continuation line). Matched with a regex, not a + prefix call, which CommentFilterAdoptionTests' census reads as a hand-rolled comment filter. */ + var isComment = Regex.IsMatch(trimmed, @"^(//|/\*|\*)"); + if (!isComment && lines[i].Contains("DROP MATERIALIZED VIEW", StringComparison.Ordinal)) + { + sites.Add((Path.GetFileName(file), i + 1)); + } + } + } + + /* Two statements: the SQL text DropRetiredBaselineRelationSql builds (run by BOTH the superseded and the + retired pass), and the reshape's own drop. Anything else is a new drop of an aggregate. */ + Assert.True(sites.Count == 2 && sites.All(s => s.File == "TimescaleSupport.cs"), + "a product drop of a continuous aggregate appeared (or moved) - give it QuiesceContinuousAggregateJobsAsync and add it here: " + + string.Join(", ", sites.Select(s => $"{s.File}:{s.Line}"))); + + var source = File.ReadAllText(Path.Combine(root, "Darling", "PerformanceMonitor.Darling.Storage", "TimescaleSupport.cs")).Replace("\r\n", "\n"); + var runners = new[] + { + "new NpgsqlCommand(DropRetiredBaselineRelationSql(legacy), connection)", + "new NpgsqlCommand(DropRetiredBaselineRelationSql(view), connection)", + "new NpgsqlCommand($\"DROP MATERIALIZED VIEW IF EXISTS collect.{view} CASCADE\", connection)", + }; + foreach (var runner in runners) + { + var at = source.IndexOf(runner, StringComparison.Ordinal); + Assert.True(at > 0, $"the drop statement '{runner}' moved"); + Assert.Equal(at, source.LastIndexOf(runner, StringComparison.Ordinal)); + var window = source[Math.Max(0, at - 900)..at]; + Assert.Matches(@"QuiesceContinuousAggregateJobsAsync\(connection, \w+, logger, quiesce, cancellationToken\);\s+if \(hold\.Skip\)\s+\{\s+continue;\s+\}", window); + /* ...and the drop is followed by the hold being cleared (the jobs went with the aggregate). */ + var after = source[at..Math.Min(source.Length, at + 400)]; + Assert.Contains("hold = AggregateJobHold.None;", after, StringComparison.Ordinal); + } + + /* The three catches reopen and start the jobs again (one helper, which names the stopped jobs when the + reopen fails), and the three cancellation catches start them again on the resume's own grace token. */ + Assert.Equal(3, Regex.Matches(source, + @"if \(!await ReopenAndResumeAfterFailedDropAsync\(\s*connection, hold, \w+, logger,\s*""[^""]*"", cancellationToken\)\)\s+\{\s+return dropped;\s+\}").Count); + Assert.Equal(3, Regex.Matches(source, + @"catch \(OperationCanceledException\) when \(hold\.Unscheduled\.Count > 0\)\s+\{\s+(/\*.*?\*/\s+)?await ResumeContinuousAggregateJobsAsync\(connection, hold\.Unscheduled, \w+, logger\);\s+throw;", + RegexOptions.Singleline).Count); + } + + /// + /// The quiesce's own exits (#5551 review): the stop is INSIDE the try whose catch schedules the jobs again, the + /// resume has no caller token to be cancelled with, and the cap decides on the final look. + /// + [Fact] + public void Quiesce_StopIsInsideTheTry_ResumeRunsOnItsOwnGrace_AndTheCapDecidesOnTheFinalLook() + { + var source = File.ReadAllText(Path.Combine(RepoRoot(), "Darling", "PerformanceMonitor.Darling.Storage", "TimescaleSupport.cs")).Replace("\r\n", "\n"); + var start = source.IndexOf("internal static async Task QuiesceContinuousAggregateJobsAsync(", StringComparison.Ordinal); + Assert.True(start > 0); + var end = source.IndexOf("/// The longest a resume of stopped jobs may take", start, StringComparison.Ordinal); + Assert.True(end > start); + var body = source[start..end]; + + Assert.Matches(@"try\s+\{\s+if \(stoppedIds\.Length > 0\)\s+\{\s+using var stop = new NpgsqlCommand\(UnscheduleJobsSql, connection\)", body); + Assert.Matches(@"catch \(Exception\)\s+\{\s+(/\*[^*]*\*/\s+)?await ResumeContinuousAggregateJobsAsync\(connection, stoppedIds, view, logger\);\s+throw;", body); + Assert.Matches(@"if \(clock\.Elapsed >= options\.Cap\)\s+\{\s+if \(workers == 0\)\s+\{\s+return hold;\s+\}", body); + + /* No resume anywhere in the product takes a caller token. */ + Assert.DoesNotContain("ResumeContinuousAggregateJobsAfterCancellationAsync", source, StringComparison.Ordinal); + Assert.Matches(@"internal static async Task ResumeContinuousAggregateJobsAsync\(\s+NpgsqlConnection connection,\s+IReadOnlyList stoppedJobIds,\s+string view,\s+ILogger\? logger\)", source); + + /* A resume that cannot happen names the jobs and the view. */ + Assert.Equal(4, Regex.Matches(source, @"StoppedJobsNote\(").Count); // the definition + the three log lines that use it + + /* Round 2: the caller's reopen on the grace is inside a try that takes the grace's OperationCanceledException, + names the jobs, and lets it through only when the caller's own token is cancelled. */ + Assert.Matches(@"try\s+\{\s+using var grace = new CancellationTokenSource\(resumeGrace \?\? s_resumeGrace\);\s+reopened = await ReopenBrokenConnectionAsync\(connection, logger, grace\.Token, reopenFailed\);\s+\}\s+catch \(OperationCanceledException\)\s+\{\s+logger\?\.LogWarning\(reopenFailed,[^;]+;\s+if \(cancellationToken\.IsCancellationRequested\)\s+\{\s+throw;", source); + Assert.Equal(2, Regex.Matches(source, @"new CancellationTokenSource\((resumeGrace \?\? )?s_resumeGrace\)").Count); // the resume's and the reopen's: the only two grace tokens + } + + private static string RepoRoot([CallerFilePath] string thisFile = "") + { + var dir = Path.GetDirectoryName(thisFile)!; + while (dir is not null && !File.Exists(Path.Combine(dir, "Directory.Build.props"))) + { + dir = Path.GetDirectoryName(dir); + } + + Assert.False(dir is null, "could not locate the repo root from the test source path"); + return dir!; + } +} + +/// +/// #5551 review round 2: the reopen after a failed drop runs on the resume grace, so a reopen that outlasts the grace +/// must end the sweep with the stopped jobs named, like any failed reopen, and must never reach the caller as a +/// cancellation the caller did not make. No database: the connection is never opened and the grace is zero, so the +/// reopen's token is cancelled before it starts. +/// +[Trait("Stage", "Guard")] +public sealed class AggregateJobGraceReopenTests +{ + private const string ReopenFailed = "The retired-baseline sweep's connection broke and could not be reopened, so the rest of the sweep is skipped until the next pass retries it: {Message}"; + + private static NpgsqlConnection ClosedConnection() + => new("Host=127.0.0.1;Port=1;Username=nobody;Database=none;Timeout=1"); + + [Fact] + public async Task ReopenThatOutlastsTheGrace_EndsTheSweepAndNamesTheStoppedJobs_NotACancellation() + { + var log = new CapturingTestLogger(); + await using var connection = ClosedConnection(); + var hold = new TimescaleSupport.AggregateJobHold(new[] { 41, 42 }, false); + + var proceed = await TimescaleSupport.ReopenAndResumeAfterFailedDropAsync( + connection, hold, "some_view", log, ReopenFailed, CancellationToken.None, TimeSpan.Zero); + + Assert.False(proceed); + Assert.Contains("did not reopen within 0 s", log.Joined); + Assert.Contains("Job(s) 41, 42 of continuous aggregate some_view are still stopped", log.Joined); + Assert.Contains("alter_job(, scheduled => true)", log.Joined); + Assert.Equal(1, log.CountAtLevel(LogLevel.Warning)); + } + + [Fact] + public async Task ReopenThatOutlastsTheGrace_WhileTheCallerIsCancelled_LogsTheJobsThenLetsTheCancellationThrough() + { + var log = new CapturingTestLogger(); + await using var connection = ClosedConnection(); + var hold = new TimescaleSupport.AggregateJobHold(new[] { 7 }, false); + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => TimescaleSupport.ReopenAndResumeAfterFailedDropAsync( + connection, hold, "some_view", log, ReopenFailed, cts.Token, TimeSpan.Zero)); + + Assert.Contains("Job(s) 7 of continuous aggregate some_view are still stopped", log.Joined); + } +} + +/// +/// #5551, live: the sweeps against a real TimescaleDB scheduler. +/// +/// #1776 own-store, not [Collection("live-postgres")]: the tests run a real job worker, hold a +/// session advisory lock the worker waits on, plant event triggers and read every job of the database, none of which +/// can share the darlingtest store with other classes' aggregates and policies. +/// +/// The worker is made to run deterministically: the aggregate selects through an IMMUTABLE-labelled +/// function that takes a shared advisory lock, and the test holds the exclusive one, so the refresh job the scheduler +/// launches sits in pg_stat_activity as Refresh Continuous Aggregate Policy [id] until the test +/// lets it go. The scheduler is left running (nothing here calls stop_background_workers); every policy starts a +/// day out so none fires on its own. +/// +public sealed class AggregateJobQuiesceLiveTests +{ + private const long GateKey = 5551; + private const string Retired = "cpu_utilization_baseline"; + + private static readonly TimescaleSupport.AggregateJobQuiesceOptions FastCap = new(TimeSpan.FromSeconds(2), TimeSpan.FromMilliseconds(100)); + + /// One look and no wait: the skip path of (c) runs before the scheduler can end the worker. + private static readonly TimescaleSupport.AggregateJobQuiesceOptions ZeroCap = new(TimeSpan.Zero, TimeSpan.FromMilliseconds(50)); + + /// (a) The jobs are stopped before the drop, and the drop still removes the aggregate and its jobs. + [Fact] + public async Task Sweep_StopsTheAggregatesJobsBeforeTheDrop_AndTheDropStillRemovesThemAll_AgainstDevPostgres() + { + var ct = TestContext.Current.CancellationToken; + var baseConnectionString = Environment.GetEnvironmentVariable("DARLING_TEST_PG"); + Assert.SkipWhen(string.IsNullOrEmpty(baseConnectionString), "Set DARLING_TEST_PG to a Postgres connection string (with TimescaleDB installed) to run the live #5551 quiesce tests."); + await using var scratch = await ScratchPostgres.CreateAsync(baseConnectionString!, ct); + await using var connection = await OpenStoreAsync(scratch, ct); + await CreateRetiredAggregateAsync(connection, Retired, ct); + await PlantDropProbeAsync(connection, Retired, ct); + + var jobIds = await JobIdsAsync(connection, Retired, ct); + Assert.Equal(3, jobIds.Count); + Assert.All(await JobStatesAsync(connection, jobIds, ct), s => Assert.True(s.Value, "every job starts scheduled")); + + var log = new CapturingTestLogger(); + Assert.Equal(1, await TimescaleSupport.DropRetiredBaselineAggregatesAsync(connection, log, DateTime.UtcNow, FastCap, ct)); + + /* The probe ran at ddl_command_start of the DROP: every job was already stopped by then. */ + Assert.Equal(3L, await ScalarAsync(connection, "SELECT count(DISTINCT job_id) FROM collect.w5551_seen", ct)); + Assert.Equal(0L, await ScalarAsync(connection, "SELECT count(*) FROM collect.w5551_seen WHERE scheduled", ct)); + Assert.False(await RelationExistsAsync(connection, Retired, ct), "the aggregate is gone"); + Assert.Equal(0L, await ScalarAsync(connection, $"SELECT count(*) FROM timescaledb_information.jobs WHERE hypertable_name = '{Retired}'", ct)); + Assert.Equal(0, log.CountAtLevel(LogLevel.Warning)); + } + + /// + /// (b) A running job worker holds the drop until it exits. Real worker: the gate keeps it inside the refresh. + /// While it runs the jobs are stopped (the scheduler cannot start another), the aggregate is still there and the + /// sweep is still waiting; once the worker exits the drop goes ahead, with no warning. + /// + [Fact] + public async Task RunningJobWorker_HoldsTheDrop_UntilItExits_AgainstDevPostgres() + { + var ct = TestContext.Current.CancellationToken; + var baseConnectionString = Environment.GetEnvironmentVariable("DARLING_TEST_PG"); + Assert.SkipWhen(string.IsNullOrEmpty(baseConnectionString), "Set DARLING_TEST_PG to a Postgres connection string (with TimescaleDB installed) to run the live #5551 quiesce tests."); + await using var scratch = await ScratchPostgres.CreateAsync(baseConnectionString!, ct); + await using var connection = await OpenStoreAsync(scratch, ct); + await CreateRetiredAggregateAsync(connection, Retired, ct); + var jobIds = await JobIdsAsync(connection, Retired, ct); + var refreshJob = await RefreshJobIdAsync(connection, Retired, ct); + + await using var gate = new NpgsqlConnection(scratch.ConnectionString); + await gate.OpenAsync(ct); + await ExecuteAsync(gate, $"SELECT pg_advisory_lock({GateKey})", ct); + await using var sweeper = new NpgsqlConnection(scratch.ConnectionString); + await sweeper.OpenAsync(ct); + Task? sweep = null; + var gateOpen = false; + try + { + await StartRefreshWorkerAsync(connection, refreshJob, ct); + Assert.True(await WaitForAsync(async () => await WorkerCountAsync(connection, jobIds, ct) > 0, TimeSpan.FromSeconds(60), ct), + "the scheduler never started the refresh worker"); + + var log = new CapturingTestLogger(); + sweep = Task.Run(() => TimescaleSupport.DropRetiredBaselineAggregatesAsync( + sweeper, log, DateTime.UtcNow, new TimescaleSupport.AggregateJobQuiesceOptions(TimeSpan.FromSeconds(120), TimeSpan.FromMilliseconds(100)), ct), ct); + + var workerPid = await ScalarAsync(connection, $"SELECT pid FROM pg_stat_activity WHERE datname = current_database() AND backend_type LIKE '% [{refreshJob}]'", ct); + + /* The jobs are stopped while the worker runs... */ + await WaitForJobsStoppedAsync(connection, jobIds, log, ct); + + /* ...and for as long as THAT worker is alive the sweeper's own backend must never be running the DROP. + That is what proves the wait: with the wait removed the DROP would sit in the sweeper's + pg_stat_activity.query (blocked on the worker's locks, or running) from the moment the jobs are + stopped, while a relation-exists check alone passes either way, because a DROP blocked behind the worker + has not removed the view yet. Each look reads the worker, then the sweeper's query, then the worker + again, and judges only when the same worker was alive on both sides. The statement text is matched by + its start (DO $do$) as well as DROP MATERIALIZED VIEW, because track_activity_query_size cuts the long + DO block short. TimescaleDB's own scheduler may end a worker whose job was just stopped (measured at + about 3-4 s): then nothing is left to hold the sweep, and the test only passes if it OBSERVED the hold + first - at least one look with the worker alive and the sweeper inside its wait loop (its last + statement is the worker count). Looked at for two seconds. */ + var sweeperPid = sweeper.ProcessID; + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(2); + var observedHold = false; + while (DateTime.UtcNow < deadline) + { + var existsBefore = await RelationExistsAsync(connection, Retired, ct); + var aliveBefore = await WorkerAliveAsync(connection, workerPid, refreshJob, ct); + var sweeperQuery = await ScalarAsync(connection, $"SELECT coalesce(query, '') FROM pg_stat_activity WHERE pid = {sweeperPid}", ct); + var aliveAfter = await WorkerAliveAsync(connection, workerPid, refreshJob, ct); + if (!aliveBefore || !aliveAfter) + { + break; + } + + Assert.True(existsBefore, $"the aggregate was dropped under a running job worker: {log.Joined}"); + Assert.False( + sweeperQuery.Contains("DROP MATERIALIZED VIEW", StringComparison.Ordinal) || sweeperQuery.TrimStart().StartsWith("DO $do$", StringComparison.Ordinal), + $"the sweeper reached its DROP while the job worker (pid {workerPid}) was still alive: {log.Joined}"); + Assert.False(sweep.IsCompleted, $"the sweep finished while a worker was running: {log.Joined}"); + observedHold |= sweeperQuery.Contains("pg_stat_activity", StringComparison.Ordinal); + await Task.Delay(TimeSpan.FromMilliseconds(50), ct); + } + + if (!observedHold) + { + Assert.Skip($"TimescaleDB ended the job worker before the hold could be observed, so this run proves nothing about the wait: {log.Joined}"); + } + + await ExecuteAsync(gate, $"SELECT pg_advisory_unlock({GateKey})", ct); + gateOpen = true; + + Assert.Equal(1, await sweep.WaitAsync(TimeSpan.FromSeconds(60), ct)); + Assert.False(await RelationExistsAsync(connection, Retired, ct), "the drop goes ahead once the worker has exited"); + Assert.Equal(0, log.CountAtLevel(LogLevel.Warning)); + } + finally + { + if (!gateOpen) + { + await ExecuteAsync(gate, $"SELECT pg_advisory_unlock({GateKey})", CancellationToken.None); + } + + if (sweep is not null) + { + try { await sweep.WaitAsync(TimeSpan.FromSeconds(60), CancellationToken.None); } + catch (Exception) { /* the assertions above already carry the failure */ } + } + } + } + + /// + /// (c) At the cap, with the worker still running at the final look: the jobs are scheduled again, the aggregate + /// stays, the sweep ends no backend itself, one warning names it; the next pass, with the worker gone, drops it. + /// + [Fact] + public async Task AtTheCap_TheJobsAreScheduledAgain_TheAggregateStays_AndTheNextPassDropsIt_AgainstDevPostgres() + { + var ct = TestContext.Current.CancellationToken; + var baseConnectionString = Environment.GetEnvironmentVariable("DARLING_TEST_PG"); + Assert.SkipWhen(string.IsNullOrEmpty(baseConnectionString), "Set DARLING_TEST_PG to a Postgres connection string (with TimescaleDB installed) to run the live #5551 quiesce tests."); + await using var scratch = await ScratchPostgres.CreateAsync(baseConnectionString!, ct); + await using var connection = await OpenStoreAsync(scratch, ct); + await CreateRetiredAggregateAsync(connection, Retired, ct); + var jobIds = await JobIdsAsync(connection, Retired, ct); + var refreshJob = await RefreshJobIdAsync(connection, Retired, ct); + + await using var gate = new NpgsqlConnection(scratch.ConnectionString); + await gate.OpenAsync(ct); + await ExecuteAsync(gate, $"SELECT pg_advisory_lock({GateKey})", ct); + var gateOpen = false; + try + { + await StartRefreshWorkerAsync(connection, refreshJob, ct); + Assert.True(await WaitForAsync(async () => await WorkerCountAsync(connection, jobIds, ct) > 0, TimeSpan.FromSeconds(60), ct), + "the scheduler never started the refresh worker"); + var workerPid = await ScalarAsync(connection, $"SELECT pid FROM pg_stat_activity WHERE datname = current_database() AND backend_type LIKE '% [{refreshJob}]'", ct); + + var log = new CapturingTestLogger(); + /* A zero cap: the first look sees the worker, so the skip and the resume follow within milliseconds of the + stop. TimescaleDB's scheduler ends a worker whose job was stopped within a few seconds (measured at + 3-4 s), and a cap of seconds could let that land inside the wait, which would turn this into a drop. */ + Assert.Equal(0, await TimescaleSupport.DropRetiredBaselineAggregatesAsync(connection, log, DateTime.UtcNow, ZeroCap, ct)); + + Assert.True(await RelationExistsAsync(connection, Retired, ct), "the aggregate stays at the cap"); + Assert.All(await JobStatesAsync(connection, jobIds, ct), s => Assert.True(s.Value, "every job it stopped is scheduled again")); + Assert.Equal(1, log.CountAtLevel(LogLevel.Warning)); + Assert.Contains(log.Lines, l => l.StartsWith("Warning:", StringComparison.Ordinal) && l.Contains(Retired, StringComparison.Ordinal) && l.Contains("#5551", StringComparison.Ordinal)); + /* The sweep itself ended nothing: the very same backend is still running. */ + Assert.Equal(1L, await ScalarAsync(connection, $"SELECT count(*) FROM pg_stat_activity WHERE pid = {workerPid} AND backend_type LIKE '% [{refreshJob}]'", ct)); + + await ExecuteAsync(gate, $"SELECT pg_advisory_unlock({GateKey})", ct); + gateOpen = true; + Assert.True(await WaitForAsync(async () => await WorkerCountAsync(connection, jobIds, ct) == 0, TimeSpan.FromSeconds(60), ct), "the worker finishes once released"); + + var retry = new CapturingTestLogger(); + Assert.Equal(1, await TimescaleSupport.DropRetiredBaselineAggregatesAsync(connection, retry, DateTime.UtcNow, FastCap, ct)); + Assert.False(await RelationExistsAsync(connection, Retired, ct), "the next pass drops it"); + Assert.Equal(0, retry.CountAtLevel(LogLevel.Warning)); + } + finally + { + if (!gateOpen) + { + await ExecuteAsync(gate, $"SELECT pg_advisory_unlock({GateKey})", CancellationToken.None); + } + } + } + + /// + /// A cancellation (shutdown, or the hourly pass's budget) while the sweep waits for a worker: the exception goes + /// through, and the jobs the sweep stopped are scheduled again anyway, because the resume runs on its own grace + /// token and not the cancelled one. + /// + [Fact] + public async Task CancelledWhileWaiting_SchedulesTheJobsAgain_OnATokenOfItsOwn_AgainstDevPostgres() + { + var ct = TestContext.Current.CancellationToken; + var baseConnectionString = Environment.GetEnvironmentVariable("DARLING_TEST_PG"); + Assert.SkipWhen(string.IsNullOrEmpty(baseConnectionString), "Set DARLING_TEST_PG to a Postgres connection string (with TimescaleDB installed) to run the live #5551 quiesce tests."); + await using var scratch = await ScratchPostgres.CreateAsync(baseConnectionString!, ct); + await using var connection = await OpenStoreAsync(scratch, ct); + await CreateRetiredAggregateAsync(connection, Retired, ct); + var jobIds = await JobIdsAsync(connection, Retired, ct); + var refreshJob = await RefreshJobIdAsync(connection, Retired, ct); + + await using var gate = new NpgsqlConnection(scratch.ConnectionString); + await gate.OpenAsync(ct); + await ExecuteAsync(gate, $"SELECT pg_advisory_lock({GateKey})", ct); + await using var sweeper = new NpgsqlConnection(scratch.ConnectionString); + await sweeper.OpenAsync(ct); + using var cancel = CancellationTokenSource.CreateLinkedTokenSource(ct); + try + { + await StartRefreshWorkerAsync(connection, refreshJob, ct); + Assert.True(await WaitForAsync(async () => await WorkerCountAsync(connection, jobIds, ct) > 0, TimeSpan.FromSeconds(60), ct), + "the scheduler never started the refresh worker"); + + var log = new CapturingTestLogger(); + var sweep = Task.Run(() => TimescaleSupport.DropRetiredBaselineAggregatesAsync( + sweeper, log, DateTime.UtcNow, new TimescaleSupport.AggregateJobQuiesceOptions(TimeSpan.FromSeconds(120), TimeSpan.FromMilliseconds(100)), cancel.Token)); + await WaitForJobsStoppedAsync(connection, jobIds, log, ct); + + cancel.Cancel(); + await Assert.ThrowsAnyAsync(() => sweep.WaitAsync(TimeSpan.FromSeconds(60), ct)); + + Assert.True(await RelationExistsAsync(connection, Retired, ct), "a cancelled sweep drops nothing"); + Assert.All(await JobStatesAsync(connection, jobIds, ct), s => Assert.True(s.Value, $"job {s.Key} is scheduled again after the cancellation")); + Assert.Equal(0, log.CountAtLevel(LogLevel.Warning)); + } + finally + { + await ExecuteAsync(gate, $"SELECT pg_advisory_unlock({GateKey})", CancellationToken.None); + } + } + + /// + /// A cap of zero is one look: with no worker running the final look is empty, and an empty final look lets the + /// drop go ahead (it is not skipped, and it logs no warning). + /// + [Fact] + public async Task EmptyFinalLookAtTheCap_LetsTheDropGoAhead_AgainstDevPostgres() + { + var ct = TestContext.Current.CancellationToken; + var baseConnectionString = Environment.GetEnvironmentVariable("DARLING_TEST_PG"); + Assert.SkipWhen(string.IsNullOrEmpty(baseConnectionString), "Set DARLING_TEST_PG to a Postgres connection string (with TimescaleDB installed) to run the live #5551 quiesce tests."); + await using var scratch = await ScratchPostgres.CreateAsync(baseConnectionString!, ct); + await using var connection = await OpenStoreAsync(scratch, ct); + await CreateRetiredAggregateAsync(connection, Retired, ct); + + var log = new CapturingTestLogger(); + Assert.Equal(1, await TimescaleSupport.DropRetiredBaselineAggregatesAsync(connection, log, DateTime.UtcNow, ZeroCap, ct)); + Assert.False(await RelationExistsAsync(connection, Retired, ct), "the drop goes ahead on an empty final look"); + Assert.Equal(0, log.CountAtLevel(LogLevel.Warning)); + } + + /// + /// (d) and (e): a drop that fails (planted XX000, which also breaks the connection, as the real race does) + /// schedules the jobs again after the reopen; a job that was already stopped before the sweep stays stopped. + /// + [Fact] + public async Task FailedDrop_SchedulesTheJobsAgain_ButNeverOneThatWasAlreadyStopped_AgainstDevPostgres() + { + var ct = TestContext.Current.CancellationToken; + var baseConnectionString = Environment.GetEnvironmentVariable("DARLING_TEST_PG"); + Assert.SkipWhen(string.IsNullOrEmpty(baseConnectionString), "Set DARLING_TEST_PG to a Postgres connection string (with TimescaleDB installed) to run the live #5551 quiesce tests."); + await using var scratch = await ScratchPostgres.CreateAsync(baseConnectionString!, ct); + await using var connection = await OpenStoreAsync(scratch, ct); + await CreateRetiredAggregateAsync(connection, Retired, ct); + var jobIds = await JobIdsAsync(connection, Retired, ct); + var refreshJob = await RefreshJobIdAsync(connection, Retired, ct); + + /* (e) stopped by hand before the sweep. */ + await ExecuteAsync(connection, $"SELECT alter_job({refreshJob}, scheduled => false)", ct); + await PlantFailingDropAsync(connection, Retired, ct); + + var log = new CapturingTestLogger(); + Assert.Equal(0, await TimescaleSupport.DropRetiredBaselineAggregatesAsync(connection, log, DateTime.UtcNow, FastCap, ct)); + + Assert.True(connection.State == System.Data.ConnectionState.Open, $"the connection is {connection.State}: {log.Joined}"); + Assert.True(await RelationExistsAsync(connection, Retired, ct), "the failed drop leaves the aggregate"); + var states = await JobStatesAsync(connection, jobIds, ct); + Assert.False(states[refreshJob], "the job that was stopped before the sweep stays stopped"); + Assert.All(states.Where(s => s.Key != refreshJob), s => Assert.True(s.Value, $"job {s.Key} (stopped by the sweep) is scheduled again")); + Assert.Equal(1, log.CountAtLevel(LogLevel.Warning)); + Assert.Contains(log.Lines, l => l.StartsWith("Warning:", StringComparison.Ordinal) && l.Contains($"retired baseline relation {Retired}", StringComparison.Ordinal)); + } + + /// + /// The reshape: the daily tier built on the aggregate is dropped by the CASCADE, so ITS jobs are stopped too + /// before the drop; a failed drop schedules both tiers' jobs again; a clean drop removes both tiers. + /// + [Fact] + public async Task Reshape_StopsTheDependentTiersJobsToo_AndSchedulesBothAgainWhenTheDropFails_AgainstDevPostgres() + { + var ct = TestContext.Current.CancellationToken; + var baseConnectionString = Environment.GetEnvironmentVariable("DARLING_TEST_PG"); + Assert.SkipWhen(string.IsNullOrEmpty(baseConnectionString), "Set DARLING_TEST_PG to a Postgres connection string (with TimescaleDB installed) to run the live #5551 quiesce tests."); + await using var scratch = await ScratchPostgres.CreateAsync(baseConnectionString!, ct); + await using var connection = await OpenStoreAsync(scratch, ct); + + /* The OLD query_store_stats_hourly shape (it still has a query_id column) with a daily tier on top. */ + await ExecuteAsync(connection, "CREATE TABLE collect.w5551_qs (collection_time timestamp NOT NULL, query_id bigint NOT NULL, v double precision NOT NULL)", ct); + await ExecuteAsync(connection, "SELECT create_hypertable('collect.w5551_qs', by_range('collection_time', INTERVAL '1 day'))", ct); + await ExecuteAsync(connection, @" +CREATE MATERIALIZED VIEW collect.query_store_stats_hourly WITH (timescaledb.continuous, timescaledb.materialized_only = false) AS +SELECT query_id, time_bucket('1 hour', collection_time) AS bucket, sum(v) AS s FROM collect.w5551_qs GROUP BY 1, 2 WITH NO DATA", ct); + await ExecuteAsync(connection, @" +CREATE MATERIALIZED VIEW collect.query_store_stats_daily WITH (timescaledb.continuous, timescaledb.materialized_only = false) AS +SELECT query_id, time_bucket('1 day', bucket) AS bucket, sum(s) AS s FROM collect.query_store_stats_hourly GROUP BY 1, 2 WITH NO DATA", ct); + await ExecuteAsync(connection, "SELECT add_continuous_aggregate_policy('collect.query_store_stats_hourly', start_offset => INTERVAL '3 days', end_offset => INTERVAL '1 hour', schedule_interval => INTERVAL '1 hour', initial_start => now() + INTERVAL '1 day')", ct); + await ExecuteAsync(connection, "SELECT add_continuous_aggregate_policy('collect.query_store_stats_daily', start_offset => INTERVAL '30 days', end_offset => INTERVAL '1 day', schedule_interval => INTERVAL '1 day', initial_start => now() + INTERVAL '1 day')", ct); + + var hourlyJobs = await JobIdsAsync(connection, "query_store_stats_hourly", ct); + var dailyJobs = await JobIdsAsync(connection, "query_store_stats_daily", ct); + Assert.Single(hourlyJobs); + Assert.Single(dailyJobs); + var both = hourlyJobs.Concat(dailyJobs).ToArray(); + + /* The lookup for the hourly aggregate finds the daily tier's job as well. */ + await using (var find = new NpgsqlCommand(TimescaleSupport.ContinuousAggregateJobsSql, connection)) + { + find.Parameters.AddWithValue("query_store_stats_hourly"); + await using var reader = await find.ExecuteReaderAsync(ct); + var found = new List(); + while (await reader.ReadAsync(ct)) + { + found.Add(reader.GetInt32(0)); + } + + Assert.Equal(both.OrderBy(i => i), found); + } + + /* A failed drop first: both tiers keep refreshing, and the connection works for the next view. */ + await PlantFailingDropAsync(connection, "query_store_stats_hourly", ct); + var failedLog = new CapturingTestLogger(); + Assert.Equal(0, await TimescaleSupport.DropStaleContinuousAggregatesAsync(connection, failedLog, FastCap, ct)); + Assert.True(connection.State == System.Data.ConnectionState.Open, $"the connection is {connection.State}: {failedLog.Joined}"); + Assert.True(await RelationExistsAsync(connection, "query_store_stats_hourly", ct)); + Assert.True(await RelationExistsAsync(connection, "query_store_stats_daily", ct)); + Assert.All(await JobStatesAsync(connection, both, ct), s => Assert.True(s.Value, $"job {s.Key} is scheduled again")); + Assert.Equal(1, failedLog.CountAtLevel(LogLevel.Warning)); + + /* Then a clean drop, observed at ddl_command_start: both tiers' jobs were already stopped, and both go. */ + await ExecuteAsync(connection, "DROP EVENT TRIGGER w5551_break_drop", ct); + await PlantDropProbeAsync(connection, "query_store_stats_hourly", ct); + var log = new CapturingTestLogger(); + Assert.Equal(1, await TimescaleSupport.DropStaleContinuousAggregatesAsync(connection, log, FastCap, ct)); + Assert.Equal(2L, await ScalarAsync(connection, "SELECT count(DISTINCT job_id) FROM collect.w5551_seen", ct)); + Assert.Equal(0L, await ScalarAsync(connection, "SELECT count(*) FROM collect.w5551_seen WHERE scheduled", ct)); + Assert.False(await RelationExistsAsync(connection, "query_store_stats_hourly", ct)); + Assert.False(await RelationExistsAsync(connection, "query_store_stats_daily", ct), "the CASCADE took the daily tier"); + Assert.Equal(0L, await ScalarAsync(connection, "SELECT count(*) FROM timescaledb_information.jobs WHERE hypertable_name IN ('query_store_stats_hourly', 'query_store_stats_daily')", ct)); + Assert.Equal(0, log.CountAtLevel(LogLevel.Warning)); + } + + /// + /// A plain view on a store with TimescaleDB, and a store with no TimescaleDB at all, have no jobs: the quiesce + /// changes nothing, waits for nothing, and the drop is exactly what it was. + /// + [Fact] + public async Task PlainViewAndPlainPostgresStore_AreNoOps_AgainstDevPostgres() + { + var ct = TestContext.Current.CancellationToken; + var baseConnectionString = Environment.GetEnvironmentVariable("DARLING_TEST_PG"); + Assert.SkipWhen(string.IsNullOrEmpty(baseConnectionString), "Set DARLING_TEST_PG to a Postgres connection string (with TimescaleDB installed) to run the live #5551 quiesce tests."); + + /* No extension: the information views do not exist, and the lookup is never run. */ + await using (var plain = await ScratchPostgres.CreateAsync(baseConnectionString!, ct)) + await using (var plainConnection = new NpgsqlConnection(plain.ConnectionString)) + { + await plainConnection.OpenAsync(ct); + Assert.False(await ScalarAsync(plainConnection, TimescaleSupport.ContinuousAggregateJobsProbeSql, ct)); + var hold = await TimescaleSupport.QuiesceContinuousAggregateJobsAsync(plainConnection, Retired, null, FastCap, ct); + Assert.Empty(hold.Unscheduled); + Assert.False(hold.Skip); + } + + /* The extension, but the relation is a plain view (the fallback branch) or absent. */ + await using var scratch = await ScratchPostgres.CreateAsync(baseConnectionString!, ct); + await using var connection = await OpenStoreAsync(scratch, ct); + await ExecuteAsync(connection, "CREATE OR REPLACE VIEW collect.file_io_baseline AS SELECT 1 AS bucket", ct); + var viewHold = await TimescaleSupport.QuiesceContinuousAggregateJobsAsync(connection, "file_io_baseline", null, FastCap, ct); + Assert.Empty(viewHold.Unscheduled); + Assert.False(viewHold.Skip); + var absentHold = await TimescaleSupport.QuiesceContinuousAggregateJobsAsync(connection, "no_such_aggregate", null, FastCap, ct); + Assert.Empty(absentHold.Unscheduled); + Assert.False(absentHold.Skip); + + var log = new CapturingTestLogger(); + Assert.Equal(1, await TimescaleSupport.DropRetiredBaselineAggregatesAsync(connection, log, DateTime.UtcNow, FastCap, ct)); + Assert.False(await RelationExistsAsync(connection, "file_io_baseline", ct), "the plain view still drops"); + Assert.Equal(0, log.CountAtLevel(LogLevel.Warning)); + } + + private static async Task OpenStoreAsync(ScratchPostgres scratch, CancellationToken ct) + { + var connection = new NpgsqlConnection(scratch.ConnectionString); + var handedOver = false; + try + { + await connection.OpenAsync(ct); + await PgMigrations.MigrateAsync(connection, ct); + + /* #1922: CREATE EXTENSION can terminate the backend it runs on when timescaledb is on disk but not + preloaded, so it runs on the probe's own connection, never on the one this method hands back. + The store is migrated first, so the extension lands in `collect`, as the product's does. */ + var timescaleEnabled = await LiveTimescaleProbe.TryEnableAsync(scratch.ConnectionString, ct); + Assert.SkipWhen(!timescaleEnabled, "The live #5551 quiesce tests need TimescaleDB."); + + handedOver = true; + return connection; + } + finally + { + if (!handedOver) + { + /* A skip or a failed setup must not leak the connection: the caller never got it to dispose. */ + await connection.DisposeAsync(); + } + } + } + + /// The pre-#2007 retired aggregate with a refresh, a compression and a retention job, none due for a day. + private static async Task CreateRetiredAggregateAsync(NpgsqlConnection connection, string view, CancellationToken ct) + { + await ExecuteAsync(connection, "CREATE TABLE IF NOT EXISTS collect.w5551_src (collection_time timestamp NOT NULL, v double precision NOT NULL)", ct); + await ExecuteAsync(connection, "SELECT create_hypertable('collect.w5551_src', by_range('collection_time', INTERVAL '1 day'), if_not_exists => true)", ct); + await ExecuteAsync(connection, "INSERT INTO collect.w5551_src VALUES (((now() AT TIME ZONE 'UTC') - INTERVAL '2 hours'), 1)", ct); + await ExecuteAsync(connection, @" +CREATE OR REPLACE FUNCTION collect.w5551_gate(x double precision) RETURNS double precision LANGUAGE plpgsql IMMUTABLE AS $f$ +BEGIN + PERFORM pg_advisory_lock_shared(5551); + PERFORM pg_advisory_unlock_shared(5551); + RETURN x; +END +$f$", ct); + await ExecuteAsync(connection, $@" +CREATE MATERIALIZED VIEW collect.{view} WITH (timescaledb.continuous, timescaledb.materialized_only = false) AS +SELECT time_bucket('1 hour', collection_time) AS bucket, sum(collect.w5551_gate(v)) AS s FROM collect.w5551_src GROUP BY 1 WITH NO DATA", ct); + await ExecuteAsync(connection, $"SELECT add_continuous_aggregate_policy('collect.{view}', start_offset => INTERVAL '3 days', end_offset => INTERVAL '1 hour', schedule_interval => INTERVAL '1 hour', initial_start => now() + INTERVAL '1 day')", ct); + await ExecuteAsync(connection, $"ALTER MATERIALIZED VIEW collect.{view} SET (timescaledb.compress = true)", ct); + await ExecuteAsync(connection, $"SELECT add_compression_policy('collect.{view}', INTERVAL '7 days', initial_start => now() + INTERVAL '1 day')", ct); + await ExecuteAsync(connection, $"SELECT add_retention_policy('collect.{view}', drop_after => INTERVAL '30 days', initial_start => now() + INTERVAL '1 day')", ct); + } + + /// Records, at the start of the DROP, whether each of the view's jobs was scheduled. TimescaleDB runs the + /// drop of an aggregate as DROP VIEW (measured), so the trigger listens for both tags, and it can fire more + /// than once per drop - the assertions are over every row it wrote. + private static async Task PlantDropProbeAsync(NpgsqlConnection connection, string view, CancellationToken ct) + { + await ExecuteAsync(connection, "CREATE TABLE collect.w5551_seen (job_id integer, scheduled boolean)", ct); + await ExecuteAsync(connection, $@" +CREATE OR REPLACE FUNCTION collect.w5551_probe() RETURNS event_trigger LANGUAGE plpgsql AS $fn$ +BEGIN + INSERT INTO collect.w5551_seen + SELECT j.job_id, j.scheduled FROM timescaledb_information.jobs AS j + WHERE j.hypertable_schema = 'collect' AND j.hypertable_name IN ('{view}', 'query_store_stats_daily') + ORDER BY j.job_id; +END +$fn$", ct); + await ExecuteAsync(connection, "CREATE EVENT TRIGGER w5551_probe ON ddl_command_start WHEN TAG IN ('DROP VIEW', 'DROP MATERIALIZED VIEW') EXECUTE FUNCTION collect.w5551_probe()", ct); + } + + /// Raises XX000 from inside the view's DROP: the error, and the broken connection, the refresh race produces. + private static async Task PlantFailingDropAsync(NpgsqlConnection connection, string view, CancellationToken ct) + { + await ExecuteAsync(connection, $@" +CREATE OR REPLACE FUNCTION collect.w5551_break_drop() RETURNS event_trigger LANGUAGE plpgsql AS $fn$ +BEGIN + IF EXISTS (SELECT 1 FROM pg_event_trigger_dropped_objects() WHERE object_name = '{view}') + THEN + RAISE EXCEPTION 'planted #5551 drop failure' USING ERRCODE = 'XX000'; + END IF; +END +$fn$", ct); + await ExecuteAsync(connection, "CREATE EVENT TRIGGER w5551_break_drop ON sql_drop EXECUTE FUNCTION collect.w5551_break_drop()", ct); + } + + private static async Task StartRefreshWorkerAsync(NpgsqlConnection connection, int refreshJob, CancellationToken ct) + => await ExecuteAsync(connection, $"SELECT alter_job({refreshJob}, next_start => now(), scheduled => true)", ct); + + private static async Task> JobIdsAsync(NpgsqlConnection connection, string view, CancellationToken ct) + { + var ids = new List(); + await using var command = new NpgsqlCommand("SELECT job_id FROM timescaledb_information.jobs WHERE hypertable_schema = 'collect' AND hypertable_name = $1 ORDER BY job_id", connection); + command.Parameters.AddWithValue(view); + await using var reader = await command.ExecuteReaderAsync(ct); + while (await reader.ReadAsync(ct)) + { + ids.Add(reader.GetInt32(0)); + } + + return ids; + } + + private static async Task RefreshJobIdAsync(NpgsqlConnection connection, string view, CancellationToken ct) + => await ScalarAsync(connection, $"SELECT job_id FROM timescaledb_information.jobs WHERE proc_name = 'policy_refresh_continuous_aggregate' AND hypertable_schema = 'collect' AND hypertable_name = '{view}'", ct); + + private static async Task> JobStatesAsync(NpgsqlConnection connection, IReadOnlyCollection jobIds, CancellationToken ct) + { + var states = new Dictionary(); + await using var command = new NpgsqlCommand("SELECT job_id, scheduled FROM timescaledb_information.jobs WHERE job_id = ANY($1::integer[])", connection); + command.Parameters.AddWithValue(jobIds.ToArray()); + await using var reader = await command.ExecuteReaderAsync(ct); + while (await reader.ReadAsync(ct)) + { + states[reader.GetInt32(0)] = reader.GetBoolean(1); + } + + Assert.Equal(jobIds.Count, states.Count); + return states; + } + + /// Waits until the sweep has stopped every one of the jobs. A drop that ran under the worker deletes the + /// jobs instead, and that is reported as what it is rather than as a missing row. + private static async Task WaitForJobsStoppedAsync(NpgsqlConnection connection, IReadOnlyCollection jobIds, CapturingTestLogger log, CancellationToken ct) + { + async Task<(long Present, long Scheduled)> LookAsync() + { + await using var command = new NpgsqlCommand( + "SELECT count(*), count(*) FILTER (WHERE scheduled) FROM timescaledb_information.jobs WHERE job_id = ANY($1::integer[])", connection); + command.Parameters.AddWithValue(jobIds.ToArray()); + await using var reader = await command.ExecuteReaderAsync(ct); + Assert.True(await reader.ReadAsync(ct)); + return (reader.GetInt64(0), reader.GetInt64(1)); + } + + var reached = await WaitForAsync(async () => { var look = await LookAsync(); return look.Present == 0 || look.Scheduled == 0; }, TimeSpan.FromSeconds(30), ct); + var final = await LookAsync(); + Assert.True(reached, $"the sweep never stopped the jobs: {log.Joined}"); + Assert.True(final.Present == jobIds.Count, $"the aggregate's jobs were dropped before its worker exited: {log.Joined}"); + } + + private static async Task WorkerAliveAsync(NpgsqlConnection connection, int workerPid, int jobId, CancellationToken ct) + => await ScalarAsync(connection, $"SELECT count(*) FROM pg_stat_activity WHERE pid = {workerPid} AND backend_type LIKE '% [{jobId}]'", ct) > 0; + + private static async Task WorkerCountAsync(NpgsqlConnection connection, IReadOnlyCollection jobIds, CancellationToken ct) + { + await using var command = new NpgsqlCommand(TimescaleSupport.AggregateJobWorkerCountSql, connection); + command.Parameters.AddWithValue(jobIds.ToArray()); + return (long)(await command.ExecuteScalarAsync(ct))!; + } + + private static async Task RelationExistsAsync(NpgsqlConnection connection, string view, CancellationToken ct) + => await ScalarAsync(connection, $"SELECT to_regclass('collect.{view}') IS NOT NULL", ct); + + private static async Task WaitForAsync(Func> condition, TimeSpan timeout, CancellationToken ct) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (await condition()) + { + return true; + } + + await Task.Delay(TimeSpan.FromMilliseconds(100), ct); + } + + return await condition(); + } + + private static async Task ScalarAsync(NpgsqlConnection connection, string sql, CancellationToken ct) + { + await using var command = new NpgsqlCommand(sql, connection) { CommandTimeout = 120 }; + var value = await command.ExecuteScalarAsync(ct); + Assert.NotNull(value); + return (T)Convert.ChangeType(value, typeof(T), System.Globalization.CultureInfo.InvariantCulture)!; + } + + private static async Task ExecuteAsync(NpgsqlConnection connection, string sql, CancellationToken ct) + { + await using var command = new NpgsqlCommand(sql, connection) { CommandTimeout = 120 }; + await command.ExecuteNonQueryAsync(ct); + } +} diff --git a/Darling/PerformanceMonitor.Darling.Service/DarlingSelfAlertEvaluator.cs b/Darling/PerformanceMonitor.Darling.Service/DarlingSelfAlertEvaluator.cs index 307e99f541..2da657471d 100644 --- a/Darling/PerformanceMonitor.Darling.Service/DarlingSelfAlertEvaluator.cs +++ b/Darling/PerformanceMonitor.Darling.Service/DarlingSelfAlertEvaluator.cs @@ -7725,6 +7725,13 @@ await RecordResolutionAsync(new AlertResolution( /// it cannot produce a delta; and if one ever did, the sentence to say about it belongs to the hold, not /// to this arm. /// + /// One known benign source (#5551). The startup and hourly sweeps that drop a retired or reshaped + /// continuous aggregate stop its jobs first, and TimescaleDB's scheduler ends a worker that is running at that + /// moment, which records ONE failed run (total_failures + 1, last run Failed). When the drop goes ahead + /// the job and its statistics are deleted with the aggregate and this arm never sees it. When the drop is + /// skipped (the worker outlasted the sweep's cap) or fails, the job stays, scheduled again, and carries that + /// failure until its next run succeeds, so this arm can report it once at INFORMATION. + /// /// Event-shaped, so there is no standing state and no resolution row: what is reported is "N more /// failures since the previous sample", which is true when it is said and is not a condition that later /// clears. The alert cooldown still gates the re-fire, so a job failing every few minutes cannot outrun diff --git a/Darling/PerformanceMonitor.Darling.Storage/TimescaleSupport.cs b/Darling/PerformanceMonitor.Darling.Storage/TimescaleSupport.cs index 22d0ae6822..4cdcbcedd1 100644 --- a/Darling/PerformanceMonitor.Darling.Storage/TimescaleSupport.cs +++ b/Darling/PerformanceMonitor.Darling.Storage/TimescaleSupport.cs @@ -1251,18 +1251,37 @@ public static Task DropRetiredBaselineAggregatesAsync( /// reasoning: now()::timestamp renders in the session zone, /// bucket is naive UTC, and PostgreSQL would compare the two without a word). /// - public static async Task DropRetiredBaselineAggregatesAsync( + public static Task DropRetiredBaselineAggregatesAsync( NpgsqlConnection connection, ILogger? logger, DateTime utcNow, CancellationToken cancellationToken = default) + => DropRetiredBaselineAggregatesAsync(connection, logger, utcNow, AggregateJobQuiesceOptions.Default, cancellationToken); + + /// + /// The same sweep with the wait for a running job worker (#5551) set by the caller, which is how the live tests + /// reach the cap in a second instead of 's thirty. Before each + /// drop of a continuous aggregate the sweep stops that aggregate's jobs and waits for a worker already running + /// (); if the worker outlasts the cap the aggregate is skipped + /// this pass, and if the drop fails its jobs are scheduled again. A plain-PostgreSQL store and a plain view have + /// no jobs, so this changes nothing for them. + /// + public static async Task DropRetiredBaselineAggregatesAsync( + NpgsqlConnection connection, ILogger? logger, DateTime utcNow, AggregateJobQuiesceOptions quiesce, + CancellationToken cancellationToken = default) { if (connection is null) { throw new ArgumentNullException(nameof(connection)); } + if (quiesce is null) + { + throw new ArgumentNullException(nameof(quiesce)); + } + var dropped = 0; foreach (var (legacy, successor) in SupersededBaselineRelations) { + var hold = AggregateJobHold.None; try { var verdict = await JudgeSupersededBaselineRelationAsync(connection, legacy, successor, utcNow, cancellationToken); @@ -1278,11 +1297,19 @@ public static async Task DropRetiredBaselineAggregatesAsync( continue; } + /* #5551: stop the aggregate's jobs, and wait out a refresh already running, BEFORE the drop. */ + hold = await QuiesceContinuousAggregateJobsAsync(connection, legacy, logger, quiesce, cancellationToken); + if (hold.Skip) + { + continue; + } + using (var drop = new NpgsqlCommand(DropRetiredBaselineRelationSql(legacy), connection) { CommandTimeout = SetupTimeoutSeconds }) { await drop.ExecuteNonQueryAsync(cancellationToken); } + hold = AggregateJobHold.None; dropped++; logger?.LogInformation( "Dropped superseded baseline relation {Legacy} (#3653) — {Reason}, so nothing reads it any more; its refresh, retention and compression jobs went with it.", @@ -1293,12 +1320,24 @@ public static async Task DropRetiredBaselineAggregatesAsync( : "it held no rows through its own view (#4289) — an empty aggregate has no baseline history to protect, whatever its successor holds" : $"its interval-honest successor {successor} exists and, like this plain view, reads raw directly"); } + catch (OperationCanceledException) when (hold.Unscheduled.Count > 0) + { + /* Shutdown or the pass's budget while this aggregate's jobs are stopped: start them again (on a + short grace of its own, not the cancelled token), then let the cancellation through as before. */ + await ResumeContinuousAggregateJobsAsync(connection, hold.Unscheduled, legacy, logger); + throw; + } catch (Exception ex) when (ex is not OperationCanceledException) { logger?.LogWarning( "Could not judge or drop superseded baseline relation {Legacy} — it lingers (harmlessly, but materializing) until the next restart retries: {Message}", legacy, ex.Message); - if (!await ReopenBrokenConnectionAsync(connection, logger, cancellationToken)) + + /* #5551: the aggregate stays, so its jobs go back to running. After the reopen, because a drop + that broke the connection is the usual way to get here; if the reopen fails the log names them. */ + if (!await ReopenAndResumeAfterFailedDropAsync( + connection, hold, legacy, logger, + "The retired-baseline sweep's connection broke and could not be reopened, so the rest of the sweep is skipped until the next pass retries it: {Message}", cancellationToken)) { return dropped; } @@ -1307,6 +1346,7 @@ public static async Task DropRetiredBaselineAggregatesAsync( foreach (var view in RetiredBaselineRelations) { + var hold = AggregateJobHold.None; try { bool existed; @@ -1320,22 +1360,37 @@ public static async Task DropRetiredBaselineAggregatesAsync( continue; } + /* #5551: stop the aggregate's jobs, and wait out a refresh already running, BEFORE the drop. */ + hold = await QuiesceContinuousAggregateJobsAsync(connection, view, logger, quiesce, cancellationToken); + if (hold.Skip) + { + continue; + } + using (var drop = new NpgsqlCommand(DropRetiredBaselineRelationSql(view), connection) { CommandTimeout = SetupTimeoutSeconds }) { await drop.ExecuteNonQueryAsync(cancellationToken); } + hold = AggregateJobHold.None; dropped++; logger?.LogInformation( "Dropped retired baseline relation {View} (#2007) — the CPU/IO anomaly arms read the raw hypertables, so nothing consumed it.", view); } + catch (OperationCanceledException) when (hold.Unscheduled.Count > 0) + { + await ResumeContinuousAggregateJobsAsync(connection, hold.Unscheduled, view, logger); + throw; + } catch (Exception ex) when (ex is not OperationCanceledException) { logger?.LogWarning( "Could not drop retired baseline relation {View} — it lingers (harmlessly, but materializing) until the next restart retries: {Message}", view, ex.Message); - if (!await ReopenBrokenConnectionAsync(connection, logger, cancellationToken)) + if (!await ReopenAndResumeAfterFailedDropAsync( + connection, hold, view, logger, + "The retired-baseline sweep's connection broke and could not be reopened, so the rest of the sweep is skipped until the next pass retries it: {Message}", cancellationToken)) { return dropped; } @@ -1345,6 +1400,352 @@ public static async Task DropRetiredBaselineAggregatesAsync( return dropped; } + /// + /// #5551: how long a drop of a continuous aggregate waits for the aggregate's job workers to exit, and how + /// often it looks. See for the numbers and why they are what they are. + /// + /// The longest the drop waits for a running worker. At the cap, if the final look still lists + /// a worker, the aggregate's jobs are scheduled again and the drop is skipped; the next pass retries it. A final + /// look that lists none lets the drop go ahead. A zero cap is one look. + /// The gap between two looks at pg_stat_activity. + public sealed record AggregateJobQuiesceOptions(TimeSpan Cap, TimeSpan PollInterval) + { + /// + /// The production values: a 30 second cap, polled every 250 ms. + /// + /// Why 30 seconds. Both sweeps run on the start path, serially, before the collectors start + /// (nothing collects while they wait), and again on the hourly store-maintenance pass, whose whole budget is + /// 5 minutes for every step (DarlingWorker's s_storeObjectConvergenceBudget). A drop happens once + /// per store, on the pass that retires or reshapes that aggregate, and only a refresh that is ALREADY running + /// when the jobs are stopped is waited for (stopping the jobs first means no new one starts). TimescaleDB's + /// scheduler usually ends such a worker itself within a few seconds of the stop (measured: about 3-4 s; + /// the worker's refresh rolls back), and one that finishes first is just as good, so the wait is a few + /// seconds for the baseline and reshape aggregates, which are small or only a day or two old. 30 seconds + /// covers that with room, and it is a cap a start can afford. The worst case is every + /// relation of both sweeps (two superseded, two retired, four reshaped = eight) stuck behind a running + /// worker, 8 x 30 s = 4 minutes, which still fits inside the hourly pass's 5. A longer cap would let a stuck + /// worker stall startup, or eat the hourly budget; a refresh that really does take longer costs nothing + /// to wait out on the next pass, because the aggregate is skipped, not dropped under a worker. + /// + /// 250 ms is the gap between the two empty looks that end the wait: a worker the scheduler started + /// just before the jobs were stopped can still be starting up and not yet listed on the first look. + /// + public static readonly AggregateJobQuiesceOptions Default = new(TimeSpan.FromSeconds(30), TimeSpan.FromMilliseconds(250)); + } + + /// + /// #5551: what did to one aggregate's jobs: + /// is the ids it stopped (the ones that were scheduled; a job that was already + /// stopped is never in it, so it is never started again), and says the drop must not run + /// this pass because a worker was still running at the cap (the jobs are already scheduled again). + /// + internal readonly record struct AggregateJobHold(IReadOnlyList Unscheduled, bool Skip) + { + /// Nothing was stopped and nothing blocks the drop: no TimescaleDB, a plain view, or no jobs. + public static AggregateJobHold None => new(Array.Empty(), false); + } + + /// Whether this store has the TimescaleDB information views at all. A plain-PostgreSQL store does not, + /// and a statement naming them fails to parse there (42P01), so the jobs are only looked up when this + /// says so (the same to_regclass probe the drop statements themselves use). + public const string ContinuousAggregateJobsProbeSql = + "SELECT to_regclass('timescaledb_information.continuous_aggregates') IS NOT NULL AND to_regclass('timescaledb_information.jobs') IS NOT NULL"; + + /// + /// #5551: every job of one continuous aggregate in collect, AND of every aggregate built on it (a + /// hierarchical daily tier), because DROP MATERIALIZED VIEW ... CASCADE drops those too and a refresh of + /// the daily one running under the CASCADE is the same race. $1 the view name, bound as a parameter. The + /// join matches EITHER identity a job can report, exactly as + /// does (timescaledb_information.jobs reports a continuous aggregate's refresh, compression and + /// retention jobs under the user view; the materialization hypertable is the other name a job can carry), and + /// the two are disjoint, so a job is listed once. A dependent aggregate reports the PARENT's materialization + /// hypertable as its source (hypertable_name), which is how the recursive part finds it. Returns the + /// job id and whether it is scheduled NOW; no row for a plain view or an absent relation. + /// + public const string ContinuousAggregateJobsSql = @" +WITH RECURSIVE aggregates AS ( + SELECT ca.view_schema, ca.view_name, + ca.materialization_hypertable_schema AS mat_schema, + ca.materialization_hypertable_name AS mat_name + FROM timescaledb_information.continuous_aggregates AS ca + WHERE ca.view_schema = 'collect' + AND ca.view_name = $1::text + UNION + SELECT child.view_schema, child.view_name, + child.materialization_hypertable_schema, + child.materialization_hypertable_name + FROM timescaledb_information.continuous_aggregates AS child + JOIN aggregates AS parent + ON child.hypertable_schema = parent.mat_schema + AND child.hypertable_name = parent.mat_name +) +SELECT DISTINCT j.job_id, j.scheduled +FROM timescaledb_information.jobs AS j +JOIN aggregates AS a + ON (a.view_schema = j.hypertable_schema AND a.view_name = j.hypertable_name) + OR (a.mat_schema = j.hypertable_schema AND a.mat_name = j.hypertable_name) +ORDER BY j.job_id"; + + /// + /// #5551: stops jobs, by id. $1 an integer[] of job ids (alter_job takes INTEGER, the + /// #1586 trap), read back through timescaledb_information.jobs so an id that vanished in the meantime is + /// skipped instead of raising. One statement, so the jobs stop together or not at all. Only scheduled is + /// named: every other alter_job argument left out means "unchanged". + /// + public const string UnscheduleJobsSql = + "SELECT alter_job(j.job_id, scheduled => false) FROM timescaledb_information.jobs AS j WHERE j.job_id = ANY($1::integer[])"; + + /// #5551: the inverse of , given only the ids that statement stopped. + public const string RescheduleJobsSql = + "SELECT alter_job(j.job_id, scheduled => true) FROM timescaledb_information.jobs AS j WHERE j.job_id = ANY($1::integer[])"; + + /// + /// #5551: how many backends of THIS database are running one of the given jobs. A TimescaleDB job worker's + /// backend_type is the job's type and its id, e.g. Refresh Continuous Aggregate Policy [1232], so + /// the id is matched as the trailing " [id]" (the type text varies by job, the id does not). Job ids are + /// per database, hence current_database(). $1 an integer[]. + /// + public const string AggregateJobWorkerCountSql = @" +SELECT count(*) +FROM pg_stat_activity AS a +WHERE a.datname = current_database() +AND EXISTS (SELECT 1 FROM unnest($1::integer[]) AS id WHERE a.backend_type LIKE '% [' || id::text || ']')"; + + /// + /// #5551: makes the drop of one continuous aggregate safe from its own jobs, before the drop runs. A refresh + /// (or compression, or retention) worker running when DROP MATERIALIZED VIEW ... CASCADE lands makes the + /// drop fail with XX000: tuple concurrently deleted (#5416), and is the inferred cause of a PostgreSQL + /// backend crash seen in CI (#5549). The same order the test harness uses before it drops a scratch database: + /// + /// find the aggregate's jobs (); + /// stop the ones that are scheduled now () and remember which; + /// a job already stopped is left stopped, and never started again; + /// wait until two looks in a row () show no worker + /// of ANY of the aggregate's jobs, stopped or not, for at most ; + /// at the cap, if the final look still lists a worker: schedule the stopped jobs again, warn, + /// and return so the caller leaves this aggregate alone until the next + /// pass. A final look that lists none lets the drop go ahead. + /// + /// This method does not cancel or terminate a backend itself, but stopping a job makes TimescaleDB's + /// scheduler end that job's running worker (measured on TimescaleDB 2.30.1: within a few seconds of the + /// stop). The refresh then rolls back, which is harmless to the data, the server log records the worker's error + /// ("job N threw an error"), and the job's statistics record one failed run (total_failures + 1, + /// last_run_status = Failed). After a successful drop the job and its statistics row are gone with the + /// aggregate; on the skip and failed-drop paths the job stays, scheduled again, with that failure on its record + /// until its next run succeeds (the #3816 job-failure alert can report it once; see its doc). So the wait is + /// for the worker to exit, whether the scheduler ended it or it finished. + /// A plain-PostgreSQL store, a plain view and an absent relation have no jobs, so this changes nothing and + /// returns . If the stop itself or anything after it throws (including a + /// cancellation, and a stop that committed but lost its connection) the stopped jobs are scheduled again, on a + /// grace of their own, before it propagates; if even that fails the log names the jobs and the view. So an + /// aggregate is not left stopped by this method without a log line that says which jobs. + /// The CALLER owns the jobs once it gets a hold back: it starts them again when the drop fails + /// (), and after a successful drop they are gone with the aggregate. + /// + internal static async Task QuiesceContinuousAggregateJobsAsync( + NpgsqlConnection connection, + string view, + ILogger? logger, + AggregateJobQuiesceOptions options, + CancellationToken cancellationToken) + { + bool hasInformationViews; + using (var probe = new NpgsqlCommand(ContinuousAggregateJobsProbeSql, connection) { CommandTimeout = SetupTimeoutSeconds }) + { + hasInformationViews = await probe.ExecuteScalarAsync(cancellationToken) is true; + } + + if (!hasInformationViews) + { + return AggregateJobHold.None; + } + + var jobs = new List<(int JobId, bool Scheduled)>(); + using (var find = new NpgsqlCommand(ContinuousAggregateJobsSql, connection) { CommandTimeout = SetupTimeoutSeconds }) + { + find.Parameters.AddWithValue(view); + using var reader = await find.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + { + jobs.Add((reader.GetInt32(0), reader.GetBoolean(1))); + } + } + + if (jobs.Count == 0) + { + return AggregateJobHold.None; + } + + var allIds = jobs.Select(j => j.JobId).ToArray(); + var stoppedIds = jobs.Where(j => j.Scheduled).Select(j => j.JobId).ToArray(); + var hold = new AggregateJobHold(stoppedIds, false); + + /* The stop is INSIDE the try: a stop that committed on the server but then lost its connection (or timed + out) still has to be undone, and scheduling an id that was never stopped is a no-op. */ + try + { + if (stoppedIds.Length > 0) + { + using var stop = new NpgsqlCommand(UnscheduleJobsSql, connection) { CommandTimeout = SetupTimeoutSeconds }; + stop.Parameters.AddWithValue(stoppedIds); + await stop.ExecuteNonQueryAsync(cancellationToken); + } + + /* Two empty looks in a row: a worker the scheduler launched just before the stop can still be starting + up and not yet listed on the first look. The deadline is checked BETWEEN looks, never by cancelling a + statement in flight, so a cap cannot break the connection it is about to use again. At the cap the + FINAL look decides: a worker still listed there skips the drop, an empty one lets it go ahead. */ + var clock = Stopwatch.StartNew(); + var emptyLooks = 0; + while (true) + { + long workers; + using (var count = new NpgsqlCommand(AggregateJobWorkerCountSql, connection) { CommandTimeout = SetupTimeoutSeconds }) + { + count.Parameters.AddWithValue(allIds); + workers = (long)(await count.ExecuteScalarAsync(cancellationToken))!; + } + + emptyLooks = workers == 0 ? emptyLooks + 1 : 0; + if (emptyLooks >= 2) + { + return hold; + } + + if (clock.Elapsed >= options.Cap) + { + if (workers == 0) + { + return hold; + } + + logger?.LogWarning( + "Not dropping continuous aggregate {View} this pass (#5551): a job worker of it was still running at the last look, {Seconds:0.0} s after its jobs were stopped (TimescaleDB usually ends such a worker itself, and this one had not exited), so the jobs are scheduled again and the drop is retried on the next pass. A worker the stop ended, if any, is recorded as one failed run in its job's statistics.", + view, clock.Elapsed.TotalSeconds); + await ResumeContinuousAggregateJobsAsync(connection, stoppedIds, view, logger); + return new AggregateJobHold(Array.Empty(), true); + } + + await Task.Delay(options.PollInterval, cancellationToken); + } + } + catch (Exception) + { + /* A cancellation (shutdown, or the pass's budget) or a failure of the stop or of a look (a broken + connection, a lock timeout). The caller never gets a hold back, so the jobs this method stopped are + scheduled again here, on a grace of their own that the cancelled token cannot take away, before the + exception reaches the caller's catch. */ + await ResumeContinuousAggregateJobsAsync(connection, stoppedIds, view, logger); + throw; + } + } + + /// The longest a resume of stopped jobs may take, on its own token: shutdown or the pass's budget may + /// already have cancelled the caller's. + private static readonly TimeSpan s_resumeGrace = TimeSpan.FromSeconds(10); + + /// The operator's way back, put in every log line that says jobs stayed stopped: the ids and the view. + private static string StoppedJobsNote(IReadOnlyList jobIds, string view) + => $"Job(s) {string.Join(", ", jobIds)} of continuous aggregate {view} are still stopped; schedule each again with SELECT alter_job(, scheduled => true). A later pass reads a stopped job as one stopped by hand and leaves it alone."; + + /// + /// #5551: schedules again the jobs a drop attempt stopped, when the aggregate stays. BEST EFFORT and logged, + /// never throws: a failed drop is already being handled by the caller's catch, and an aggregate left stopped is + /// the less bad outcome than a second failure hiding the first. Runs on a grace token of its own + /// (), never the caller's, so a cancelled caller cannot swallow it. The connection is + /// reopened first if the failed statement broke it (, #5416). A job + /// the caller did not stop is never passed in, so one that was stopped by hand before the sweep stays stopped. + /// When the jobs cannot be scheduled again the log line names them and the view. + /// + internal static async Task ResumeContinuousAggregateJobsAsync( + NpgsqlConnection connection, + IReadOnlyList stoppedJobIds, + string view, + ILogger? logger) + { + if (stoppedJobIds.Count == 0) + { + return; + } + + try + { + using var grace = new CancellationTokenSource(s_resumeGrace); + if (!await ReopenBrokenConnectionAsync( + connection, logger, grace.Token, + "The connection broke while the jobs of a continuous aggregate were stopped and could not be reopened, so they could not be scheduled again: {Message} " + + StoppedJobsNote(stoppedJobIds, view))) + { + return; + } + + using var start = new NpgsqlCommand(RescheduleJobsSql, connection) { CommandTimeout = SetupTimeoutSeconds }; + start.Parameters.AddWithValue(stoppedJobIds.ToArray()); + await start.ExecuteNonQueryAsync(grace.Token); + logger?.LogInformation( + "Scheduled {Count} job(s) of continuous aggregate {View} again (#5551): the drop did not go ahead, so it keeps refreshing.", + stoppedJobIds.Count, view); + } + catch (Exception ex) + { + logger?.LogWarning( + "Could not schedule the stopped jobs of continuous aggregate {View} again after the drop did not go ahead (#5551): {Message} {Note}", + view, ex.Message, StoppedJobsNote(stoppedJobIds, view)); + } + } + + /// + /// #5551: what each sweep's per-relation catch does after a failed drop while it holds stopped jobs: reopen the + /// connection if the failure broke it, then schedule the jobs again. Returns false when the connection could not + /// be reopened (the sweep ends, as before); that log line names the stopped jobs and the view, so the operator + /// knows which to schedule again. With nothing stopped it is exactly . + /// The reopen of a sweep that holds jobs runs on the resume grace, not the caller's token, so an + /// out of it is the GRACE running out (a pool wait, a host that stopped + /// answering; Npgsql's own connect timeout is longer than the grace) and never the caller's cancellation: it is + /// logged with the stopped job ids and the view, and the sweep ends like any failed reopen. It reaches the caller + /// only when the caller's own token was cancelled too, and then after that same log line. This method never + /// throws anything else: the resume it ends with takes every exception itself. + /// is the seam for a test; production leaves it at . + /// + internal static async Task ReopenAndResumeAfterFailedDropAsync( + NpgsqlConnection connection, + AggregateJobHold hold, + string view, + ILogger? logger, + string reopenFailedTemplate, + CancellationToken cancellationToken, + TimeSpan? resumeGrace = null) + { + if (hold.Unscheduled.Count == 0) + { + return await ReopenBrokenConnectionAsync(connection, logger, cancellationToken, reopenFailedTemplate); + } + + var reopenFailed = reopenFailedTemplate + " " + StoppedJobsNote(hold.Unscheduled, view); + var reopened = false; + try + { + using var grace = new CancellationTokenSource(resumeGrace ?? s_resumeGrace); + reopened = await ReopenBrokenConnectionAsync(connection, logger, grace.Token, reopenFailed); + } + catch (OperationCanceledException) + { + logger?.LogWarning(reopenFailed, "the connection did not reopen within " + (resumeGrace ?? s_resumeGrace).TotalSeconds.ToString("0.###", System.Globalization.CultureInfo.InvariantCulture) + " s"); + if (cancellationToken.IsCancellationRequested) + { + throw; + } + } + + if (!reopened) + { + return false; + } + + await ResumeContinuousAggregateJobsAsync(connection, hold.Unscheduled, view, logger); + return true; + } + /// /// #5416: after a failed attempt, hands the sweep (and every convergence step that shares its connection /// after it) an OPEN connection again. A continuous aggregate's refresh job running at the moment of the @@ -5674,13 +6075,27 @@ always been and the statement a store without batching emits is byte-identical t /// old shape in place (logged), never kills startup. query_stats CAGGs are unchanged and untouched. CASCADE /// drops the dependent daily CAGG, which the ensure sweep also recreates. /// - public static async Task DropStaleContinuousAggregatesAsync(NpgsqlConnection connection, ILogger? logger, CancellationToken cancellationToken = default) + public static Task DropStaleContinuousAggregatesAsync(NpgsqlConnection connection, ILogger? logger, CancellationToken cancellationToken = default) + => DropStaleContinuousAggregatesAsync(connection, logger, AggregateJobQuiesceOptions.Default, cancellationToken); + + /// + /// The same reshape with the wait for a running job worker (#5551) set by the caller; see + /// . + /// The jobs of an aggregate built on the one being dropped are stopped with it, because the CASCADE takes both. + /// + public static async Task DropStaleContinuousAggregatesAsync( + NpgsqlConnection connection, ILogger? logger, AggregateJobQuiesceOptions quiesce, CancellationToken cancellationToken = default) { if (connection is null) { throw new ArgumentNullException(nameof(connection)); } + if (quiesce is null) + { + throw new ArgumentNullException(nameof(quiesce)); + } + var reshapes = new[] { /* OLD query_store_stats_hourly grouped by query_id/plan_id → stale iff it still has a query_id column. */ @@ -5706,6 +6121,7 @@ The ensure sweep recreates it WITH NO DATA and the eight-day refresh window re-m var dropped = 0; foreach (var (view, staleCheck) in reshapes) { + var hold = AggregateJobHold.None; try { bool stale; @@ -5719,21 +6135,46 @@ The ensure sweep recreates it WITH NO DATA and the eight-day refresh window re-m continue; } + /* #5551: stop the aggregate's jobs (and those of any aggregate built on it, which the CASCADE below + takes too), and wait out a refresh already running, BEFORE the drop. */ + hold = await QuiesceContinuousAggregateJobsAsync(connection, view, logger, quiesce, cancellationToken); + if (hold.Skip) + { + continue; + } + using (var drop = new NpgsqlCommand($"DROP MATERIALIZED VIEW IF EXISTS collect.{view} CASCADE", connection) { CommandTimeout = SetupTimeoutSeconds }) { await drop.ExecuteNonQueryAsync(cancellationToken); } + hold = AggregateJobHold.None; dropped++; logger?.LogInformation( "TimescaleDB: dropped stale continuous aggregate {View} (composer-dimension reshape) — recreated in the new shape this cycle.", view); } + catch (OperationCanceledException) when (hold.Unscheduled.Count > 0) + { + await ResumeContinuousAggregateJobsAsync(connection, hold.Unscheduled, view, logger); + throw; + } catch (Exception ex) when (ex is not OperationCanceledException) { logger?.LogWarning( "Reshape drop of {View} failed — it stays in the OLD shape until the next restart retries: {Message}", view, ex.Message); + + /* #5551: a drop that lost to a running refresh (XX000) closes the connection (#5416), and this loop + had no reopen, so the NEXT view's stale check threw "Connection is not open" and logged a second, + misleading failure. Reopened here the way the retired-baseline sweep does, and the stopped jobs + start again so the aggregate that stays keeps refreshing. */ + if (!await ReopenAndResumeAfterFailedDropAsync( + connection, hold, view, logger, + "The stale-aggregate reshape's connection broke and could not be reopened, so the rest of the reshape is skipped until the next pass retries it: {Message}", cancellationToken)) + { + return dropped; + } } }