using System.Diagnostics;
using System.Globalization;
using System.Text.Json.Nodes;
using Microsoft.Data.SqlClient;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests;
///
/// The single highest-value integration test for the Sql driver: the frozen-peer /
/// blackhole gate. When the database stops answering mid-poll, a reused pooled connection hangs
/// inside ExecuteReaderAsync; the driver must surface
/// within its client-side operationTimeout (a real linked-CTS cancellation) and NOT wedge the
/// poll loop. This is the exact bug class the repo shipped and fixed once in its S7 driver (async
/// reads ignored the socket read timeout → a frozen peer wedged the poll loop; see
/// S7_1500ConnectTimeoutOutageTests, which this mirrors).
/// The wall-clock bound is the whole point. operationTimeout is set well BELOW
/// commandTimeout here, deliberately inverted from the production rule, so the two backstops
/// are distinguishable: a working client-side cancellation returns at ≈ operationTimeout; if it
/// were broken and only the ADO.NET CommandTimeout server-side backstop fired, the read would
/// return at ≈ commandTimeout instead — which the upper-bound assertion fails on. A test that
/// merely asserted "eventually BadTimeout" would pass even with the client-side cancellation broken.
/// SAFETY — this gate `docker pause`s a DEDICATED container it owns, never shared infra.
/// The pause target is the compile-time constant
/// (otopcua-sql-blackhole), never anything derived from an env var, so a mis-set endpoint can
/// never aim a pause at the rig's shared always-on SQL Server (10.100.0.35,14330, which hosts
/// ConfigDb for the whole rig). As a second guardrail the gate skips loudly if its endpoint
/// resolves to the shared server's port . Bring the dedicated stack up
/// with Docker/docker-compose.yml (container otopcua-sql-blackhole, host port 14333).
/// Env-gated; offline skip is the normal outcome. With unset
/// nothing here opens a socket, shells out to docker, or waits — the suite is instant and green on the
/// macOS dev box, exactly like every other *.IntegrationTests live gate in this tree.
///
/// Operator run recipe (docker runs on the shared Linux host; drive it from this VM):
///
/// # 1. Deploy + start the DEDICATED disposable mssql (NOT the shared 14330 server):
/// lmxopcua-fix sync sql-blackhole
/// lmxopcua-fix up sql-blackhole # container otopcua-sql-blackhole on :14333, self-seeds
///
/// # 2. Point the gate at the dedicated container + tell it how to reach docker:
/// export SQL_BLACKHOLE_ENDPOINT=10.100.0.35,14333 # the dedicated container; MUST NOT be ,14330
/// export SQL_BLACKHOLE_PASSWORD=Blackhole_dev_pw1 # matches the compose SA password
/// export SQL_BLACKHOLE_DOCKER_SSH=dohertj2@10.100.0.35 # empty ⇒ run docker locally
///
/// # 3. Run just this gate:
/// dotnet test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests \
/// --filter "FullyQualifiedName~SqlBlackholeTimeoutTests"
///
/// # 4. Tear the disposable stack down:
/// lmxopcua-fix down sql-blackhole
///
///
///
[Trait("Category", "Integration")]
[Trait("Category", "Blackhole")]
public sealed class SqlBlackholeTimeoutTests
{
/// Endpoint (host,port) of the DEDICATED disposable container. Absent ⇒ offline skip.
private const string EndpointEnvVar = "SQL_BLACKHOLE_ENDPOINT";
/// SA password for the dedicated container. Required by the run.
private const string PasswordEnvVar = "SQL_BLACKHOLE_PASSWORD";
/// SQL login. Defaults to sa.
private const string UserEnvVar = "SQL_BLACKHOLE_USER";
///
/// Optional SSH target the docker pause/unpause runs through (e.g.
/// dohertj2@10.100.0.35), since docker -H ssh:// does not work from this VM. Empty ⇒
/// run docker locally.
///
private const string DockerSshEnvVar = "SQL_BLACKHOLE_DOCKER_SSH";
///
/// The ONLY container this gate ever pauses — a compile-time constant, never derived from the
/// environment. The shared central SQL Server is not named this, so no env misconfiguration can
/// cause this gate to freeze shared infra. It matches Docker/docker-compose.yml's
/// container_name.
///
private const string ContainerName = "otopcua-sql-blackhole";
///
/// The rig's shared always-on SQL Server port (hosts ConfigDb for the whole rig). An endpoint on
/// this port is refused with a loud skip — the dedicated container must be on a different port.
///
private const string SharedCentralPort = "14330";
/// The database the seed lives in; created if missing (never dropped).
private const string Database = "SqlBlackholeFixture";
/// The seeded key-value table + its columns and the one baseline key.
private const string Schema = "sqlpoll";
private const string KeyValueTable = "TagValues";
private const string KeyColumn = "tag_name";
private const string ValueColumn = "num_value";
private const string TimestampColumn = "sample_ts";
private const string BaselineKey = "Line1.Speed";
private const double BaselineValue = 42.0;
/// The RawPath the driver serves the baseline key under.
private const string RawPath = "Sql/Line1/Speed";
// ── deadlines chosen so client-side and server-side backstops are DISTINGUISHABLE ──
/// Client-side wall-clock ceiling on one group. A frozen peer must surface BadTimeout at ≈ this.
private static readonly TimeSpan OperationTimeout = TimeSpan.FromSeconds(3);
///
/// ADO.NET server-side backstop — set an order of magnitude ABOVE so
/// that if the client-side cancellation were broken and only this fired, the read would return at
/// ≈ 30 s and the upper-bound assertion would catch it.
///
private static readonly TimeSpan CommandTimeout = TimeSpan.FromSeconds(30);
///
/// Upper bound on the post-pause read. Comfortably above (docker
/// pause propagation + pooled-connection validation + thread-pool scheduling) yet comfortably below
/// — so a return here proves the CLIENT-side bound fired, not the
/// server-side backstop.
///
private static readonly TimeSpan UpperBound = TimeSpan.FromSeconds(15);
/// Hard cap on the test's own wait, so a genuinely wedged implementation FAILS rather than hangs CI.
private static readonly TimeSpan HardCap = TimeSpan.FromSeconds(60);
/// Hard cap on each docker pause/unpause shell command, so a hung SSH handshake can't hang CI.
private static readonly TimeSpan ShellCommandHardCap = TimeSpan.FromSeconds(30);
///
/// Start reading against the dedicated mssql, docker pause it mid-poll, and assert the next
/// read surfaces within (a
/// client-side linked-CTS cancellation, not the poll thread wedging and not the server-side
/// backstop), that the driver degrades (health + host
/// ), and that after docker unpause a subsequent poll recovers
/// to / .
///
[Fact]
public async Task PausedServer_surfacesBadTimeout_withinDeadline_notWedged()
{
var ep = Environment.GetEnvironmentVariable(EndpointEnvVar);
Assert.SkipWhen(
string.IsNullOrWhiteSpace(ep),
$"{EndpointEnvVar} not set — offline skip. Bring up Docker/docker-compose.yml (dedicated " +
$"container {ContainerName} on :14333) and export {EndpointEnvVar}/{PasswordEnvVar} to run this gate.");
var password = Environment.GetEnvironmentVariable(PasswordEnvVar);
Assert.SkipWhen(
string.IsNullOrWhiteSpace(password),
$"{PasswordEnvVar} not set — the blackhole gate needs the dedicated container's SA password.");
var (host, port) = ParseEndpoint(ep!);
// ── SAFETY GUARD: refuse to run against the rig's shared central SQL Server. ──
// The pause target (ContainerName) is a constant and could never be the shared container regardless,
// but refusing the shared PORT stops us even connecting-and-polling against shared infra.
Assert.SkipWhen(
string.Equals(port, SharedCentralPort, StringComparison.Ordinal),
$"{EndpointEnvVar} resolves to port {SharedCentralPort} — that is the rig's SHARED always-on SQL " +
"Server (ConfigDb for the whole rig), which this destructive gate must NEVER pause. Point it at " +
$"the dedicated {ContainerName} container (host port 14333), not the shared server.");
var user = Environment.GetEnvironmentVariable(UserEnvVar) is { Length: > 0 } named ? named : "sa";
var ssh = Environment.GetEnvironmentVariable(DockerSshEnvVar);
var ct = TestContext.Current.CancellationToken;
// Seed defensively (create-if-missing) so the baseline poll is Good even if compose never ran seed.sql.
await SeedAsync(host, port, user, password!, ct);
await using var driver = await StartDriverAsync(host, port, user, password!, ct);
// Good baseline — prove the endpoint answers before we freeze it.
var baseline = (await ReadOnceAsync(driver, ct)).ShouldHaveSingleItem();
baseline.StatusCode.ShouldBe(SqlStatusCodes.Good);
Convert.ToDouble(baseline.Value).ShouldBe(BaselineValue);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
var paused = false;
try
{
// Freeze the container mid-poll: its SQL process is SIGSTOPped, so the pooled connection's next
// ExecuteReaderAsync hangs against a live-but-silent socket — the frozen-peer wedge exactly.
// Mark paused BEFORE the await: if `ct` fires mid-command the process may still complete the pause,
// so the finally must attempt an unpause regardless — never leave the container frozen.
paused = true;
await RunShellCommandAsync(DockerCommand("pause", ssh), ct);
var clock = Stopwatch.StartNew();
// Bounded by HardCap so a wedged implementation FAILS here instead of hanging CI.
var snapshots = await ReadOnceAsync(driver, ct).WaitAsync(HardCap, ct);
clock.Stop();
var frozen = snapshots.ShouldHaveSingleItem();
frozen.StatusCode.ShouldBe(
SqlStatusCodes.BadTimeout,
"a frozen database must surface BadTimeout, not a stale Good or another Bad class");
// The wall-clock contract: returned at ≈ operationTimeout (client-side cancellation), NOT at
// commandTimeout (server-side backstop) and NOT never. UpperBound < CommandTimeout is what
// distinguishes a working client-side abort from a broken one that only the backstop rescued.
clock.Elapsed.ShouldBeLessThan(
UpperBound,
$"BadTimeout must arrive on the client-side {OperationTimeout.TotalSeconds:0}s bound, not the " +
$"server-side {CommandTimeout.TotalSeconds:0}s CommandTimeout backstop");
clock.Elapsed.ShouldBeLessThan(CommandTimeout, "the server-side backstop must not be what fired");
clock.Elapsed.ShouldBeGreaterThanOrEqualTo(
OperationTimeout - TimeSpan.FromSeconds(2),
"returning far faster than operationTimeout would mean a fast-fail misclassified as BadTimeout");
// Driver-level classification, not just the reader: an all-BadTimeout poll degrades health and
// stops the host (SqlDriver.ObservePollOutcome), even though the reader returned successfully.
driver.GetHealth().State.ShouldBe(DriverState.Degraded);
driver.GetHostStatuses().ShouldHaveSingleItem().State.ShouldBe(HostState.Stopped);
}
finally
{
if (paused)
{
// ALWAYS restore the container once a pause was attempted — never leave a frozen container
// behind. Best-effort: unpausing a container that never actually paused (pause command failed)
// errors harmlessly, and a cleanup failure must never mask the real test outcome.
try
{
await RunShellCommandAsync(DockerCommand("unpause", ssh), CancellationToken.None);
}
catch
{
// swallow — cleanup is best-effort; the container is disposable and torn down by the runbook.
}
}
}
// Recovery: after unpause a subsequent poll must return Good and the whole stack must recover.
await PollUntilGoodAsync(driver, ct);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
driver.GetHostStatuses().ShouldHaveSingleItem().State.ShouldBe(HostState.Running);
}
// ── helpers ──
/// Runs one poll of the single baseline reference.
private static Task> ReadOnceAsync(SqlDriver driver, CancellationToken ct)
=> driver.ReadAsync([RawPath], ct);
/// Polls until a Good snapshot returns, or the recovery deadline elapses.
private static async Task PollUntilGoodAsync(SqlDriver driver, CancellationToken ct)
{
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(45);
while (DateTime.UtcNow < deadline)
{
var snapshot = (await ReadOnceAsync(driver, ct)).ShouldHaveSingleItem();
if (snapshot.StatusCode == SqlStatusCodes.Good) return;
await Task.Delay(TimeSpan.FromSeconds(1), ct);
}
throw new Xunit.Sdk.XunitException(
"the driver never recovered to a Good poll within 45 s of docker unpause — the poll loop did not survive the outage");
}
/// Builds a driver over the dedicated container, warmed to hold a pooled connection.
private static async Task StartDriverAsync(
string host, string port, string user, string password, CancellationToken ct)
{
var options = new SqlDriverOptions
{
RawTags = [KeyValueTag()],
OperationTimeout = OperationTimeout,
CommandTimeout = CommandTimeout,
};
var driver = new SqlDriver(
options,
driverInstanceId: "sql-blackhole",
dialect: new SqlServerDialect(),
connectionString: DriverConnectionString(host, port, user, password));
try
{
await driver.InitializeAsync("{}", ct);
}
catch
{
await driver.DisposeAsync();
throw;
}
return driver;
}
/// The KeyValue tag over the seeded sqlpoll.TagValues baseline row.
private static RawTagEntry KeyValueTag()
{
var config = new JsonObject
{
["driver"] = "Sql",
["model"] = "KeyValue",
["table"] = $"{Schema}.{KeyValueTable}",
["keyColumn"] = KeyColumn,
["keyValue"] = BaselineKey,
["valueColumn"] = ValueColumn,
["timestampColumn"] = TimestampColumn,
};
return new RawTagEntry(RawPath, config.ToJsonString(), WriteIdempotent: false);
}
///
/// The driver's connection string. Min Pool Size=1 keeps a warm pooled connection so the
/// post-pause read reuses it and hangs inside ExecuteReaderAsync — exercising the
/// query-execution wedge (the S7 bug class) rather than a fresh connect.
///
private static string DriverConnectionString(string host, string port, string user, string password)
=> new SqlConnectionStringBuilder
{
DataSource = $"{host},{port}",
InitialCatalog = Database,
UserID = user,
Password = password,
TrustServerCertificate = true,
Encrypt = false,
ConnectTimeout = 5,
MinPoolSize = 1,
}.ConnectionString;
/// Creates the database (if missing) and (re)seeds the baseline schema + row.
private static async Task SeedAsync(
string host, string port, string user, string password, CancellationToken ct)
{
var master = new SqlConnectionStringBuilder
{
DataSource = $"{host},{port}",
InitialCatalog = "master",
UserID = user,
Password = password,
TrustServerCertificate = true,
Encrypt = false,
ConnectTimeout = 10,
}.ConnectionString;
await using (var connection = new SqlConnection(master))
{
await connection.OpenAsync(ct);
await ExecuteAsync(connection, $"IF DB_ID(N'{Database}') IS NULL CREATE DATABASE [{Database}];", ct);
}
var db = new SqlConnectionStringBuilder(master) { InitialCatalog = Database }.ConnectionString;
await using (var connection = new SqlConnection(db))
{
await connection.OpenAsync(ct);
foreach (var statement in SeedStatements())
await ExecuteAsync(connection, statement, ct);
}
}
/// The seed, one batch per statement (T-SQL requires CREATE SCHEMA to be first in its batch).
private static IEnumerable SeedStatements()
{
yield return $"IF SCHEMA_ID(N'{Schema}') IS NULL EXEC(N'CREATE SCHEMA [{Schema}]');";
yield return $"DROP TABLE IF EXISTS [{Schema}].[{KeyValueTable}];";
yield return $"""
CREATE TABLE [{Schema}].[{KeyValueTable}] (
[{KeyColumn}] nvarchar(128) NOT NULL,
[{ValueColumn}] float NULL,
[{TimestampColumn}] datetime2(3) NOT NULL
);
""";
yield return string.Create(CultureInfo.InvariantCulture, $"""
INSERT INTO [{Schema}].[{KeyValueTable}] ([{KeyColumn}], [{ValueColumn}], [{TimestampColumn}])
VALUES (N'{BaselineKey}', {BaselineValue}, '2026-07-24T10:00:00');
""");
}
/// Runs one non-query batch.
private static async Task ExecuteAsync(SqlConnection connection, string sql, CancellationToken ct)
{
await using var command = connection.CreateCommand();
command.CommandText = sql;
await command.ExecuteNonQueryAsync(ct);
}
/// Splits a host,port endpoint. A missing port is rejected — the guard needs the port.
private static (string Host, string Port) ParseEndpoint(string endpoint)
{
var parts = endpoint.Split(',', 2, StringSplitOptions.TrimEntries);
if (parts.Length != 2 || parts[1].Length == 0)
{
throw new Xunit.Sdk.XunitException(
$"{EndpointEnvVar} must be 'host,port' (e.g. 10.100.0.35,14333) so the shared-server port " +
$"guard can run — got '{endpoint}'.");
}
return (parts[0], parts[1]);
}
///
/// Builds the docker command for the OWNED container only. The verb and the constant container name
/// are the sole inputs — the target is never taken from the environment.
///
private static string DockerCommand(string verb, string? ssh)
{
var docker = $"docker {verb} {ContainerName}";
return string.IsNullOrWhiteSpace(ssh) ? docker : $"ssh {ssh} \"{docker}\"";
}
///
/// Runs a shell one-liner through the login shell, exactly as the S7 blackhole gate does — but bounded
/// by its own so a hung SSH handshake (unreachable docker host, an auth
/// prompt) fails the test rather than hanging CI indefinitely. The process is killed on timeout.
///
private static async Task RunShellCommandAsync(string command, CancellationToken ct)
{
using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
timeout.CancelAfter(ShellCommandHardCap);
using var proc = new Process
{
StartInfo = new ProcessStartInfo("/bin/sh", $"-c \"{command.Replace("\"", "\\\"")}\"")
{
RedirectStandardOutput = true,
RedirectStandardError = true,
UseShellExecute = false,
},
};
proc.Start();
try
{
await proc.WaitForExitAsync(timeout.Token);
}
catch (OperationCanceledException) when (timeout.IsCancellationRequested && !ct.IsCancellationRequested)
{
try { proc.Kill(entireProcessTree: true); } catch { /* already gone */ }
throw new TimeoutException(
$"shell command did not complete within {ShellCommandHardCap.TotalSeconds:0}s: {command}");
}
proc.ExitCode.ShouldBe(0, $"docker command failed: {await proc.StandardError.ReadToEndAsync(ct)}");
// Give the pause/unpause a moment to take effect before the next poll tick.
await Task.Delay(TimeSpan.FromSeconds(1), ct);
}
}