feat(sql): implement the design §8.1 catalog identifier-validation gate (#496)
Until now ISqlDialect.QuoteIdentifier was the SOLE defence for authored
table/column names: an identifier went from the TagConfig blob straight into a
command text, bracket-quoted. The residual risk was bounded — a hostile name is
quoted into one nonexistent object, the query fails after the connection opens,
and the tag Bad-codes — but "bounded" is not "filtered", and the design promised
a filter. Both ISqlDialect and SqlServerDialect carried a doc paragraph saying so.
SqlCatalogGate + SqlCatalogLoader now resolve every authored identifier against
the live catalog at Initialize, and REPLACE it with the catalog's own spelling.
The identifier text in an emitted poll query is therefore a string this driver
read back out of ListSchemas/ListTables/ListColumns — not operator input. Quoting
becomes the backstop it was documented to be.
Decisions worth knowing before touching this:
- Substitution, not just validation. Matching is exact-ordinal first, then a
UNIQUE case-insensitive hit: SQL Server's default collation is CI so case
variants have always worked, and rejecting them would break valid deployments.
An ambiguous CI match under a case-sensitive collation is refused rather than
guessed — picking one would publish another column's data under the operator's
node, which is worse than rejecting the tag.
- Charset check BEFORE catalog lookup. Each identifier goes through
QuoteIdentifier for its rejection rules first, so a name carrying a control or
Unicode format character is rejected WITHOUT its value being echoed into a log
line (Trojan-Source). A name that passes is safe to render, which is why
catalog-miss messages do name it — an operator hunting a typo has to see what
they wrote. Both halves are pinned by tests.
- A rejected tag keeps its node. It is dropped from the POLLED table but stays in
the AUTHORED table, so it still materializes and reads BadNodeIdUnknown —
§8.1's specified outcome. My first wiring dropped it from both, which deleted
the node instead; an existing test (ReinitializeAsync_recoversFromFaulted)
caught it. A status code can only be published by a node that exists, and a
missing address-space entry is far harder to diagnose than a Bad quality.
- An unreadable catalog FAULTS Initialize; it does not reject every tag. That is
the absence of evidence about the tags, not evidence against them — rejecting
all of them would serve a confidently-empty address space and send the operator
hunting typos that do not exist. Zero visible schemas is treated the same way,
because that is exactly what a missing GRANT looks like. Faulting lands
DriverInstanceActor in Reconnecting with its retry timer running.
- Bounded load: one schema list, one default-schema scalar, then one ListTables
per distinct authored schema and one ListColumns per distinct authored relation
— never a full catalog enumeration, and nothing after Initialize. Every
authored name reaches the catalog queries as a bound @schema/@table parameter,
so building the allow-list cannot itself be an injection vector.
- ISqlDialect gains DefaultSchemaSql ("SELECT SCHEMA_NAME()"). An unqualified
`TagValues` must resolve in whatever schema the SERVER reports; hardcoding dbo
would be a silent lie on any estate that maps service accounts to their own
default schema. It is a query, not a constant, because the answer is
per-connection.
- Accepted v1 limitation: a 3-part db.schema.table (or linked-server name)
addresses a catalog this connection cannot enumerate, so it cannot be
allow-listed and is rejected with a message pointing at the fix — expose the
data through a view in the connected database.
VerifyLivenessAsync's wall-clock pattern is extracted to RunBoundedAsync and
reused, so the new I/O gets the same R2-01 / STAB-14 protection rather than a
second hand-rolled copy. (It also fixes a latent bug in the extracted code: the
OperationCanceledException arm detached the wrong task.)
Tests: 24 pure gate tests + 9 end-to-end driver tests against the real SQLite
catalog + 3 new live tests against the real SQL Server on 10.100.0.35 (21 in that
suite now pass, exercising the real SELECT SCHEMA_NAME() + INFORMATION_SCHEMA
path). Verified falsifiable: bypassing the gate turns exactly the three rejection
assertions red and leaves the rest green.
SqlInjectionRegressionTests is deliberately NOT rewritten to expect
BadNodeIdUnknown. It drives SqlPollReader directly, below the gate, and pins the
quoting backstop on its own — defence in depth is only worth the name if each
layer holds independently. Rewriting those assertions would delete the backstop's
only coverage and leave the gate a single point of failure. Its scope note, which
said the gate does not exist, is updated to say why it stays where it is.
This commit is contained in:
Joseph Doherty
@@ -14,11 +14,13 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Sql;
|
||||
/// parameterized in SQL, so the few that must appear as text are emitted through
|
||||
/// <see cref="QuoteIdentifier"/>. Any code path that builds SQL by concatenating an authored tag field
|
||||
/// <em>without</em> going through <see cref="QuoteIdentifier"/> is a defect (design §8.1).</para>
|
||||
/// <para><b>Not yet implemented:</b> design §8.1 also specifies that an authored table/column is
|
||||
/// validated against the live catalog before it is ever quoted into text, so that an unknown identifier
|
||||
/// rejects the tag. No such gate exists yet — an identifier reaches <see cref="QuoteIdentifier"/>
|
||||
/// straight from the authored <c>TagConfig</c> blob, so <b>quoting is currently the only defence</b>,
|
||||
/// not a backstop behind an upstream filter. Scrutinise it accordingly.</para>
|
||||
/// <para><b>Quoting is the backstop, not the only defence.</b> Design §8.1's catalog gate is
|
||||
/// implemented: <see cref="SqlCatalogGate"/> resolves every authored table/column against the live
|
||||
/// catalog at Initialize and <b>replaces it with the catalog's own spelling</b>, so an identifier
|
||||
/// reaching <see cref="QuoteIdentifier"/> on the poll path is a string this driver read back out of
|
||||
/// <see cref="ListSchemasSql"/> / <see cref="ListTablesSql"/> / <see cref="ListColumnsSql"/> — not
|
||||
/// operator input. An identifier that resolves to nothing rejects its tag (which then publishes
|
||||
/// <c>BadNodeIdUnknown</c>) rather than being quoted into a query against a nonexistent object.</para>
|
||||
/// <para>Public because <c>Driver.Sql.Browser</c> consumes it — the catalog SQL <em>is</em> the browse
|
||||
/// engine, so it is shared rather than duplicated. Implementations own their provider package;
|
||||
/// <b>no provider-specific type appears in this signature</b> (<see cref="Factory"/> is the abstract
|
||||
@@ -89,6 +91,20 @@ public interface ISqlDialect
|
||||
/// <summary>Catalog query listing schemas — the browser's <c>RootAsync</c> level. Takes no parameters.</summary>
|
||||
string ListSchemasSql { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Scalar query returning the schema an <b>unqualified</b> object name resolves to for the connecting
|
||||
/// principal — T-SQL's <c>SELECT SCHEMA_NAME()</c>. Takes no parameters.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>Needed by <see cref="SqlCatalogGate"/>: a tag authored as <c>TagValues</c> rather than
|
||||
/// <c>dbo.TagValues</c> has to be looked up in <em>some</em> schema, and guessing <c>dbo</c> would be a
|
||||
/// silent lie on any estate that maps its service accounts to their own default schema. Asking the
|
||||
/// server is the only answer that matches how the poll query itself will resolve the name.</para>
|
||||
/// <para>It is a <em>query</em> rather than a name because the answer is per-connection, not per-dialect
|
||||
/// — the same dialect resolves differently for two different logins.</para>
|
||||
/// </remarks>
|
||||
string DefaultSchemaSql { get; }
|
||||
|
||||
/// <summary>Catalog query listing tables + views in one schema — the browser's schema-expand level. Bind <c>@schema</c>.</summary>
|
||||
string ListTablesSql { get; }
|
||||
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Sql;
|
||||
|
||||
/// <summary>
|
||||
/// An immutable snapshot of the parts of a database's catalog the authored tags actually name — the
|
||||
/// allow-list design §8.1 requires identifiers to come from.
|
||||
/// <para><b>This type holds catalog strings only.</b> Every name in it was read back out of
|
||||
/// <see cref="ISqlDialect.ListSchemasSql"/> / <see cref="ISqlDialect.ListTablesSql"/> /
|
||||
/// <see cref="ISqlDialect.ListColumnsSql"/>; nothing an operator typed is ever stored here. That is what
|
||||
/// lets <see cref="SqlCatalogGate"/> hand the planner catalog spellings rather than authored ones, so the
|
||||
/// identifier text in an emitted query is a string the database gave us.</para>
|
||||
/// <para><b>Deliberately partial.</b> Only the schemas/tables the authored tags name are loaded — a
|
||||
/// driver polling three tables must not enumerate a warehouse's ten thousand. A name absent from this
|
||||
/// snapshot therefore means "not found <em>when we looked</em>", which is exactly the question the gate
|
||||
/// asks.</para>
|
||||
/// </summary>
|
||||
public sealed class SqlCatalog
|
||||
{
|
||||
private readonly IReadOnlyList<string> _schemas;
|
||||
|
||||
/// <summary>Canonical schema → canonical table names in it.</summary>
|
||||
private readonly IReadOnlyDictionary<string, IReadOnlyList<string>> _tablesBySchema;
|
||||
|
||||
/// <summary>Canonical <c>schema.table</c> → that relation's canonical column names.</summary>
|
||||
private readonly IReadOnlyDictionary<string, IReadOnlyList<string>> _columnsByTable;
|
||||
|
||||
/// <summary>Constructs a catalog snapshot.</summary>
|
||||
/// <param name="defaultSchema">The schema an unqualified object name resolves to for this connection.</param>
|
||||
/// <param name="schemas">Every schema the connection can see.</param>
|
||||
/// <param name="tablesBySchema">Canonical schema → its tables/views.</param>
|
||||
/// <param name="columnsByTable">Canonical <c>schema.table</c> → its columns.</param>
|
||||
internal SqlCatalog(
|
||||
string defaultSchema,
|
||||
IReadOnlyList<string> schemas,
|
||||
IReadOnlyDictionary<string, IReadOnlyList<string>> tablesBySchema,
|
||||
IReadOnlyDictionary<string, IReadOnlyList<string>> columnsByTable)
|
||||
{
|
||||
DefaultSchema = defaultSchema;
|
||||
_schemas = schemas;
|
||||
_tablesBySchema = tablesBySchema;
|
||||
_columnsByTable = columnsByTable;
|
||||
}
|
||||
|
||||
/// <summary>The schema an unqualified authored object name is resolved in.</summary>
|
||||
public string DefaultSchema { get; }
|
||||
|
||||
/// <summary>Resolves an authored schema name to the catalog's own spelling.</summary>
|
||||
/// <param name="authored">The authored schema, or null/blank for the default schema.</param>
|
||||
/// <param name="canonical">The catalog's spelling, when resolved.</param>
|
||||
/// <returns><see langword="true"/> when the schema exists (or the default schema is used).</returns>
|
||||
public bool TryResolveSchema(string? authored, out string canonical)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(authored))
|
||||
{
|
||||
canonical = DefaultSchema;
|
||||
return true;
|
||||
}
|
||||
|
||||
return TryMatch(_schemas, authored, out canonical);
|
||||
}
|
||||
|
||||
/// <summary>Resolves an authored table/view name within a canonical schema.</summary>
|
||||
/// <param name="canonicalSchema">The schema, already resolved through <see cref="TryResolveSchema"/>.</param>
|
||||
/// <param name="authored">The authored table or view name.</param>
|
||||
/// <param name="canonical">The catalog's spelling, when resolved.</param>
|
||||
/// <returns><see langword="true"/> when the relation exists in that schema.</returns>
|
||||
public bool TryResolveTable(string canonicalSchema, string authored, out string canonical)
|
||||
{
|
||||
canonical = string.Empty;
|
||||
return _tablesBySchema.TryGetValue(canonicalSchema, out var tables)
|
||||
&& TryMatch(tables, authored, out canonical);
|
||||
}
|
||||
|
||||
/// <summary>Resolves an authored column name within a canonical relation.</summary>
|
||||
/// <param name="canonicalSchema">The resolved schema.</param>
|
||||
/// <param name="canonicalTable">The resolved table or view.</param>
|
||||
/// <param name="authored">The authored column name.</param>
|
||||
/// <param name="canonical">The catalog's spelling, when resolved.</param>
|
||||
/// <returns><see langword="true"/> when the column exists on that relation.</returns>
|
||||
public bool TryResolveColumn(
|
||||
string canonicalSchema, string canonicalTable, string authored, out string canonical)
|
||||
{
|
||||
canonical = string.Empty;
|
||||
return _columnsByTable.TryGetValue(QualifiedKey(canonicalSchema, canonicalTable), out var columns)
|
||||
&& TryMatch(columns, authored, out canonical);
|
||||
}
|
||||
|
||||
/// <summary>The dictionary key for a canonical relation.</summary>
|
||||
/// <param name="schema">The canonical schema.</param>
|
||||
/// <param name="table">The canonical table.</param>
|
||||
/// <returns>The composite key.</returns>
|
||||
internal static string QualifiedKey(string schema, string table) => schema + "." + table;
|
||||
|
||||
/// <summary>
|
||||
/// Matches an authored name against catalog spellings: an exact (ordinal) hit wins, otherwise a
|
||||
/// <b>unique</b> case-insensitive hit is accepted and its catalog spelling returned.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para><b>Why case-insensitive at all.</b> SQL Server's default collation is case-insensitive, so
|
||||
/// <c>num_value</c> and <c>NUM_VALUE</c> genuinely name the same column and both work today. Matching
|
||||
/// ordinally would reject configurations that have always been valid — a gate that breaks working
|
||||
/// deployments is not defence in depth.</para>
|
||||
/// <para><b>Why exact-first, and why ambiguity fails.</b> On a case-<em>sensitive</em> collation a
|
||||
/// relation may legitimately carry both <c>Value</c> and <c>value</c>. Preferring the exact match keeps
|
||||
/// such a config resolving to what the operator wrote, and refusing to choose between two
|
||||
/// case-insensitive candidates is the only safe answer — silently picking one would publish a different
|
||||
/// column's data under the operator's node, which is worse than rejecting the tag.</para>
|
||||
/// </remarks>
|
||||
/// <param name="candidates">The catalog spellings to match against.</param>
|
||||
/// <param name="authored">The authored name.</param>
|
||||
/// <param name="canonical">The catalog's spelling, when matched.</param>
|
||||
/// <returns><see langword="true"/> on an unambiguous match.</returns>
|
||||
internal static bool TryMatch(IReadOnlyList<string> candidates, string authored, out string canonical)
|
||||
{
|
||||
foreach (var candidate in candidates)
|
||||
{
|
||||
if (!string.Equals(candidate, authored, StringComparison.Ordinal)) continue;
|
||||
canonical = candidate;
|
||||
return true;
|
||||
}
|
||||
|
||||
canonical = string.Empty;
|
||||
var matches = 0;
|
||||
foreach (var candidate in candidates)
|
||||
{
|
||||
if (!string.Equals(candidate, authored, StringComparison.OrdinalIgnoreCase)) continue;
|
||||
if (++matches > 1)
|
||||
{
|
||||
canonical = string.Empty;
|
||||
return false;
|
||||
}
|
||||
|
||||
canonical = candidate;
|
||||
}
|
||||
|
||||
return matches == 1;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,241 @@
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Sql;
|
||||
|
||||
/// <summary>
|
||||
/// One tag the gate refused, and why. Carried rather than logged in place so the caller decides the log
|
||||
/// shape and the caller's tests can assert on the reason.
|
||||
/// </summary>
|
||||
/// <param name="RawPath">The rejected tag's identity — its RawPath, which is trusted structure, not blob content.</param>
|
||||
/// <param name="Field">The <see cref="SqlTagDefinition"/> field that failed (e.g. <c>ValueColumn</c>).</param>
|
||||
/// <param name="Reason">An operator-actionable explanation, safe to log.</param>
|
||||
public sealed record SqlCatalogRejection(string RawPath, string Field, string Reason);
|
||||
|
||||
/// <summary>The outcome of applying a <see cref="SqlCatalog"/> to a set of authored tags.</summary>
|
||||
/// <param name="Accepted">
|
||||
/// The surviving definitions, <b>rewritten to catalog spellings</b>. Safe to hand to
|
||||
/// <see cref="SqlGroupPlanner"/>.
|
||||
/// </param>
|
||||
/// <param name="Rejected">Every tag the gate refused, in input order.</param>
|
||||
public sealed record SqlCatalogGateResult(
|
||||
IReadOnlyList<SqlTagDefinition> Accepted,
|
||||
IReadOnlyList<SqlCatalogRejection> Rejected);
|
||||
|
||||
/// <summary>
|
||||
/// Design §8.1's identifier gate: validates every authored table/column against the live catalog
|
||||
/// <b>before</b> it can be quoted into SQL text, and replaces it with the catalog's own spelling.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para><b>What this closes.</b> Until this existed, <see cref="ISqlDialect.QuoteIdentifier"/> was the
|
||||
/// sole defence: an identifier went from the authored <c>TagConfig</c> blob straight into a command text,
|
||||
/// bracket-quoted. The residual risk was bounded — a hostile name is quoted into one nonexistent object,
|
||||
/// the query fails after the connection opens, and the tag Bad-codes — but "bounded" is not "filtered",
|
||||
/// and the design promised a filter. With the gate in place the identifier text in an emitted query is a
|
||||
/// string this driver read back out of the catalog, so quoting is a backstop behind an allow-list rather
|
||||
/// than the whole story.</para>
|
||||
/// <para><b>Charset first, catalog second.</b> Each identifier is passed through
|
||||
/// <see cref="ISqlDialect.QuoteIdentifier"/> before it is looked up, purely for its rejection rules
|
||||
/// (control characters, Unicode format characters, over-length). That ordering is deliberate: a name that
|
||||
/// fails the charset check is rejected <em>without its value being echoed</em>, so nothing carrying a bidi
|
||||
/// override or a NUL can reach a log line through this type's rejection messages. A name that passes has
|
||||
/// been proven safe to render, which is why the catalog-miss messages can afford to name it — and they
|
||||
/// must, or an operator has no way to find the typo.</para>
|
||||
/// <para><b>Rejection is per-tag, never driver-wide.</b> A tag naming a dropped column is dropped from the
|
||||
/// driver's table, so its RawPath resolves to nothing and it publishes
|
||||
/// <see cref="SqlStatusCodes.BadNodeIdUnknown"/> — design §8.1's specified outcome — while every other tag
|
||||
/// on that database keeps polling. This mirrors the "one malformed blob must not take the whole driver
|
||||
/// down" rule the tag-table build already follows.</para>
|
||||
/// </remarks>
|
||||
public static class SqlCatalogGate
|
||||
{
|
||||
/// <summary>
|
||||
/// The most name parts this gate can validate. A three-part <c>db.schema.table</c> (or a
|
||||
/// four-part linked-server name) addresses a catalog this connection's <c>INFORMATION_SCHEMA</c>
|
||||
/// cannot see, so it cannot be allow-listed.
|
||||
/// </summary>
|
||||
private const int MaxNameParts = 2;
|
||||
|
||||
/// <summary>
|
||||
/// Applies <paramref name="catalog"/> to <paramref name="definitions"/>, returning the accepted tags
|
||||
/// with catalog-spelled identifiers and the rejected ones with reasons.
|
||||
/// </summary>
|
||||
/// <param name="definitions">The authored definitions, as parsed from their <c>TagConfig</c> blobs.</param>
|
||||
/// <param name="catalog">The catalog snapshot to validate against.</param>
|
||||
/// <param name="dialect">Supplies the identifier charset rules via <see cref="ISqlDialect.QuoteIdentifier"/>.</param>
|
||||
/// <returns>The accepted and rejected sets.</returns>
|
||||
/// <exception cref="ArgumentNullException">A required argument is null.</exception>
|
||||
public static SqlCatalogGateResult Apply(
|
||||
IEnumerable<SqlTagDefinition> definitions, SqlCatalog catalog, ISqlDialect dialect)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(definitions);
|
||||
ArgumentNullException.ThrowIfNull(catalog);
|
||||
ArgumentNullException.ThrowIfNull(dialect);
|
||||
|
||||
var accepted = new List<SqlTagDefinition>();
|
||||
var rejected = new List<SqlCatalogRejection>();
|
||||
|
||||
foreach (var definition in definitions)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(definition);
|
||||
if (TryCanonicalize(definition, catalog, dialect, out var canonical, out var rejection))
|
||||
accepted.Add(canonical!);
|
||||
else
|
||||
rejected.Add(rejection!);
|
||||
}
|
||||
|
||||
return new SqlCatalogGateResult(accepted, rejected);
|
||||
}
|
||||
|
||||
/// <summary>Resolves one definition's identifiers, or explains the first failure.</summary>
|
||||
private static bool TryCanonicalize(
|
||||
SqlTagDefinition definition,
|
||||
SqlCatalog catalog,
|
||||
ISqlDialect dialect,
|
||||
out SqlTagDefinition? canonical,
|
||||
out SqlCatalogRejection? rejection)
|
||||
{
|
||||
canonical = null;
|
||||
rejection = null;
|
||||
|
||||
if (!TryResolveRelation(definition, catalog, dialect, out var schema, out var table, out rejection))
|
||||
return false;
|
||||
|
||||
// Every identifier-bearing field, paired with its name for the rejection message. A field the tag's
|
||||
// model does not use is null and simply skipped — the planner enforces which are required.
|
||||
var resolved = new Dictionary<string, string?>(StringComparer.Ordinal);
|
||||
foreach (var (field, authored) in new (string Field, string? Authored)[]
|
||||
{
|
||||
(nameof(SqlTagDefinition.KeyColumn), definition.KeyColumn),
|
||||
(nameof(SqlTagDefinition.ValueColumn), definition.ValueColumn),
|
||||
(nameof(SqlTagDefinition.TimestampColumn), definition.TimestampColumn),
|
||||
(nameof(SqlTagDefinition.ColumnName), definition.ColumnName),
|
||||
(nameof(SqlTagDefinition.RowSelectorColumn), definition.RowSelectorColumn),
|
||||
(nameof(SqlTagDefinition.RowSelectorTopByTimestamp), definition.RowSelectorTopByTimestamp),
|
||||
})
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(authored))
|
||||
{
|
||||
resolved[field] = authored;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!IsRenderableIdentifier(authored, dialect))
|
||||
{
|
||||
rejection = new SqlCatalogRejection(definition.Name, field,
|
||||
$"the authored {field} is not a usable identifier (it is over-long, or contains control " +
|
||||
"or Unicode format characters); the value is withheld from this message because it " +
|
||||
"cannot be safely rendered");
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!catalog.TryResolveColumn(schema, table, authored, out var column))
|
||||
{
|
||||
rejection = new SqlCatalogRejection(definition.Name, field,
|
||||
$"column '{authored}' does not exist on '{schema}.{table}' (or matches more than one " +
|
||||
"column under a case-sensitive collation)");
|
||||
return false;
|
||||
}
|
||||
|
||||
resolved[field] = column;
|
||||
}
|
||||
|
||||
canonical = definition with
|
||||
{
|
||||
Table = SqlCatalog.QualifiedKey(schema, table),
|
||||
KeyColumn = resolved[nameof(SqlTagDefinition.KeyColumn)],
|
||||
ValueColumn = resolved[nameof(SqlTagDefinition.ValueColumn)],
|
||||
TimestampColumn = resolved[nameof(SqlTagDefinition.TimestampColumn)],
|
||||
ColumnName = resolved[nameof(SqlTagDefinition.ColumnName)],
|
||||
RowSelectorColumn = resolved[nameof(SqlTagDefinition.RowSelectorColumn)],
|
||||
RowSelectorTopByTimestamp = resolved[nameof(SqlTagDefinition.RowSelectorTopByTimestamp)],
|
||||
};
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>Splits and resolves the authored <c>table</c> to a canonical schema + relation.</summary>
|
||||
private static bool TryResolveRelation(
|
||||
SqlTagDefinition definition,
|
||||
SqlCatalog catalog,
|
||||
ISqlDialect dialect,
|
||||
out string schema,
|
||||
out string table,
|
||||
out SqlCatalogRejection? rejection)
|
||||
{
|
||||
schema = string.Empty;
|
||||
table = string.Empty;
|
||||
rejection = null;
|
||||
const string Field = nameof(SqlTagDefinition.Table);
|
||||
|
||||
if (string.IsNullOrWhiteSpace(definition.Table))
|
||||
{
|
||||
rejection = new SqlCatalogRejection(definition.Name, Field,
|
||||
"the tag has no 'table'; author the table or view to read");
|
||||
return false;
|
||||
}
|
||||
|
||||
var parts = definition.Table.Split('.');
|
||||
if (parts.Length > MaxNameParts)
|
||||
{
|
||||
rejection = new SqlCatalogRejection(definition.Name, Field,
|
||||
$"'{definition.Table}' is a {parts.Length}-part name. Only 'table' and 'schema.table' can be " +
|
||||
"validated against this connection's catalog, so a cross-database or linked-server name " +
|
||||
"cannot be allow-listed; expose the data through a view in this database instead");
|
||||
return false;
|
||||
}
|
||||
|
||||
var authoredSchema = parts.Length == MaxNameParts ? parts[0] : null;
|
||||
var authoredTable = parts[^1];
|
||||
|
||||
foreach (var part in parts)
|
||||
{
|
||||
if (IsRenderableIdentifier(part, dialect)) continue;
|
||||
rejection = new SqlCatalogRejection(definition.Name, Field,
|
||||
"the authored table name has a part that is not a usable identifier (empty, over-long, or " +
|
||||
"containing control or Unicode format characters); the value is withheld from this message " +
|
||||
"because it cannot be safely rendered");
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!catalog.TryResolveSchema(authoredSchema, out schema))
|
||||
{
|
||||
rejection = new SqlCatalogRejection(definition.Name, Field,
|
||||
$"schema '{authoredSchema}' does not exist (or matches more than one schema under a " +
|
||||
"case-sensitive collation)");
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!catalog.TryResolveTable(schema, authoredTable, out table))
|
||||
{
|
||||
rejection = new SqlCatalogRejection(definition.Name, Field,
|
||||
$"table or view '{authoredTable}' does not exist in schema '{schema}'" +
|
||||
(authoredSchema is null
|
||||
? $" (the connection's default schema — qualify the name as 'schema.{authoredTable}' if it lives elsewhere)"
|
||||
: string.Empty));
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// True when the dialect's quoting rules accept <paramref name="identifier"/> — i.e. it is safe both to
|
||||
/// embed in SQL and to render into a log line or an AdminUI label.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Uses <see cref="ISqlDialect.QuoteIdentifier"/> as the charset authority rather than duplicating its
|
||||
/// rules, so the two can never disagree about what a legal identifier is. The quoted result is
|
||||
/// discarded: only whether it threw is interesting here.
|
||||
/// </remarks>
|
||||
private static bool IsRenderableIdentifier(string identifier, ISqlDialect dialect)
|
||||
{
|
||||
try
|
||||
{
|
||||
dialect.QuoteIdentifier(identifier);
|
||||
return true;
|
||||
}
|
||||
catch (ArgumentException)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,189 @@
|
||||
using System.Data;
|
||||
using System.Data.Common;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Sql;
|
||||
|
||||
/// <summary>
|
||||
/// Loads the <see cref="SqlCatalog"/> slice the authored tags need, over one connection, using only the
|
||||
/// dialect's catalog SQL (design §8.1).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para><b>Bounded by what is authored, not by what exists.</b> One query for the schema list, one for
|
||||
/// the default schema, then one <see cref="ISqlDialect.ListTablesSql"/> per distinct authored schema and
|
||||
/// one <see cref="ISqlDialect.ListColumnsSql"/> per distinct authored relation. A driver polling three
|
||||
/// tables issues a handful of round-trips at Initialize and none thereafter; it never enumerates a
|
||||
/// warehouse.</para>
|
||||
/// <para><b>Every parameter is bound.</b> Authored names reach the catalog queries as
|
||||
/// <c>@schema</c>/<c>@table</c> parameters — the same discipline the schema browser follows — so loading
|
||||
/// the allow-list cannot itself be an injection vector. No identifier is quoted into text on this path.</para>
|
||||
/// <para><b>Failures throw; they do not degrade to an empty catalog.</b> An empty catalog would reject
|
||||
/// every tag, which is indistinguishable at the OPC UA surface from a database whose objects were all
|
||||
/// dropped. A caller that cannot read the catalog has not learned that the tags are invalid — it has
|
||||
/// learned nothing — and must fail its Initialize so the driver retries rather than serving a
|
||||
/// confidently-empty address space.</para>
|
||||
/// </remarks>
|
||||
internal static class SqlCatalogLoader
|
||||
{
|
||||
/// <summary>Column alias <see cref="ISqlDialect.ListSchemasSql"/> projects.</summary>
|
||||
private const string SchemaColumn = "TABLE_SCHEMA";
|
||||
|
||||
/// <summary>Column alias <see cref="ISqlDialect.ListTablesSql"/> projects.</summary>
|
||||
private const string TableNameColumn = "TABLE_NAME";
|
||||
|
||||
/// <summary>Column alias <see cref="ISqlDialect.ListColumnsSql"/> projects.</summary>
|
||||
private const string ColumnNameColumn = "COLUMN_NAME";
|
||||
|
||||
/// <summary>
|
||||
/// Reads the catalog slice covering <paramref name="authoredTables"/>.
|
||||
/// </summary>
|
||||
/// <param name="connection">An <b>already-open</b> connection. Not disposed here; the caller owns it.</param>
|
||||
/// <param name="dialect">Supplies the catalog SQL.</param>
|
||||
/// <param name="authoredTables">The distinct authored <c>table</c> strings, as written in the blobs.</param>
|
||||
/// <param name="commandTimeout">Per-command server-side backstop.</param>
|
||||
/// <param name="cancellationToken">Cancellation token for the operation.</param>
|
||||
/// <returns>The loaded catalog.</returns>
|
||||
/// <exception cref="InvalidOperationException">The connection can see no schemas at all.</exception>
|
||||
internal static async Task<SqlCatalog> LoadAsync(
|
||||
DbConnection connection,
|
||||
ISqlDialect dialect,
|
||||
IReadOnlyCollection<string> authoredTables,
|
||||
TimeSpan commandTimeout,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var timeoutSeconds = Math.Max(1, (int)Math.Ceiling(commandTimeout.TotalSeconds));
|
||||
|
||||
var schemas = await QueryColumnAsync(
|
||||
connection, dialect.ListSchemasSql, SchemaColumn,
|
||||
static _ => { }, timeoutSeconds, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
// Zero schemas is a permissions or visibility symptom, never a real database. Rejecting every tag on
|
||||
// that basis would report an authoring fault for what is actually a grant problem, and would send the
|
||||
// operator to the wrong system — the same misdiagnosis the driver's health classification avoids.
|
||||
if (schemas.Count == 0)
|
||||
throw new InvalidOperationException(
|
||||
"the catalog returned no schemas at all; the connecting principal most likely cannot read " +
|
||||
"the catalog views, which is a grant problem rather than a tag-authoring one");
|
||||
|
||||
var defaultSchema = await ReadDefaultSchemaAsync(
|
||||
connection, dialect, timeoutSeconds, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
var tablesBySchema = new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal);
|
||||
var columnsByTable = new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal);
|
||||
|
||||
// Resolving here goes through SqlCatalog.TryMatch — the SAME matcher the gate will use — so the
|
||||
// loader and the gate can never disagree about which relation an authored name resolved to. Loading
|
||||
// one table and validating against another would be a silent wrong-column defect.
|
||||
foreach (var authored in authoredTables)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(authored)) continue;
|
||||
|
||||
var parts = authored.Split('.');
|
||||
if (parts.Length > 2) continue; // the gate rejects these by name; nothing to load.
|
||||
|
||||
var authoredTable = parts[^1];
|
||||
string schema;
|
||||
if (parts.Length == 2)
|
||||
{
|
||||
if (!SqlCatalog.TryMatch(schemas, parts[0], out schema)) continue;
|
||||
}
|
||||
else
|
||||
{
|
||||
schema = defaultSchema;
|
||||
}
|
||||
|
||||
if (!tablesBySchema.TryGetValue(schema, out var tables))
|
||||
{
|
||||
tables = await QueryColumnAsync(
|
||||
connection, dialect.ListTablesSql, TableNameColumn,
|
||||
command => Bind(command, "@schema", schema),
|
||||
timeoutSeconds, cancellationToken).ConfigureAwait(false);
|
||||
tablesBySchema[schema] = tables;
|
||||
}
|
||||
|
||||
if (!SqlCatalog.TryMatch(tables, authoredTable, out var table)) continue;
|
||||
|
||||
var key = SqlCatalog.QualifiedKey(schema, table);
|
||||
if (columnsByTable.ContainsKey(key)) continue;
|
||||
|
||||
columnsByTable[key] = await QueryColumnAsync(
|
||||
connection, dialect.ListColumnsSql, ColumnNameColumn,
|
||||
command =>
|
||||
{
|
||||
Bind(command, "@schema", schema);
|
||||
Bind(command, "@table", table);
|
||||
},
|
||||
timeoutSeconds, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return new SqlCatalog(defaultSchema, schemas, tablesBySchema, columnsByTable);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Reads the connection's default schema, falling back to the sole visible schema when the dialect's
|
||||
/// query answers null/blank.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// A null answer is possible (a login with no default schema mapped). Falling back to the single
|
||||
/// visible schema keeps the common one-schema database working; with several visible schemas there is
|
||||
/// no defensible guess, so unqualified names simply will not resolve and the gate says so by name.
|
||||
/// </remarks>
|
||||
private static async Task<string> ReadDefaultSchemaAsync(
|
||||
DbConnection connection, ISqlDialect dialect, int timeoutSeconds, CancellationToken cancellationToken)
|
||||
{
|
||||
var command = connection.CreateCommand();
|
||||
await using (command.ConfigureAwait(false))
|
||||
{
|
||||
command.CommandText = dialect.DefaultSchemaSql;
|
||||
command.CommandTimeout = timeoutSeconds;
|
||||
var value = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
|
||||
var schema = value is null or DBNull ? null : value.ToString();
|
||||
return string.IsNullOrWhiteSpace(schema) ? string.Empty : schema;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Runs one catalog query and projects a single string column from every row.</summary>
|
||||
private static async Task<IReadOnlyList<string>> QueryColumnAsync(
|
||||
DbConnection connection,
|
||||
string sql,
|
||||
string columnName,
|
||||
Action<DbCommand> bind,
|
||||
int timeoutSeconds,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var results = new List<string>();
|
||||
|
||||
var command = connection.CreateCommand();
|
||||
await using (command.ConfigureAwait(false))
|
||||
{
|
||||
command.CommandText = sql;
|
||||
command.CommandTimeout = timeoutSeconds;
|
||||
bind(command);
|
||||
|
||||
var reader = await command
|
||||
.ExecuteReaderAsync(CommandBehavior.SingleResult, cancellationToken).ConfigureAwait(false);
|
||||
await using (reader.ConfigureAwait(false))
|
||||
{
|
||||
var ordinal = -1;
|
||||
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
if (ordinal < 0) ordinal = reader.GetOrdinal(columnName);
|
||||
if (reader.IsDBNull(ordinal)) continue;
|
||||
var value = reader.GetValue(ordinal)?.ToString();
|
||||
if (!string.IsNullOrWhiteSpace(value)) results.Add(value);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return results;
|
||||
}
|
||||
|
||||
/// <summary>Binds one catalog-query parameter — the only way an authored name reaches the database here.</summary>
|
||||
private static void Bind(DbCommand command, string name, string value)
|
||||
{
|
||||
var parameter = command.CreateParameter();
|
||||
parameter.ParameterName = name;
|
||||
parameter.DbType = DbType.String;
|
||||
parameter.Value = value;
|
||||
command.Parameters.Add(parameter);
|
||||
}
|
||||
}
|
||||
@@ -74,7 +74,23 @@ public sealed class SqlDriver
|
||||
private IReadOnlyDictionary<string, SqlTagDefinition> _tagsByRawPath =
|
||||
new Dictionary<string, SqlTagDefinition>(StringComparer.Ordinal);
|
||||
|
||||
/// <summary>Resolves a read/subscribe RawPath to its authored definition — a single table hit, or a miss.</summary>
|
||||
/// <summary>
|
||||
/// The subset of <see cref="_tagsByRawPath"/> that survived the design §8.1 catalog gate, with every
|
||||
/// identifier rewritten to the catalog's own spelling. <b>This — not the authored table — is what a
|
||||
/// read or a poll resolves against</b>, so an identifier that never appeared in the catalog can never
|
||||
/// reach a query.
|
||||
/// <para><b>Why the two tables are separate.</b> A tag the gate rejected must still <em>exist</em> as a
|
||||
/// node: §8.1's specified outcome is that it publishes
|
||||
/// <see cref="SqlStatusCodes.BadNodeIdUnknown"/>, and a status code can only be published by a node
|
||||
/// that is there. Dropping rejected tags from the authored table too would delete the node instead,
|
||||
/// turning a diagnosable Bad quality into a silently missing address-space entry — a strictly worse
|
||||
/// failure for the operator trying to find their typo.</para>
|
||||
/// <para>Swapped atomically, for the same reason the authored table is.</para>
|
||||
/// </summary>
|
||||
private IReadOnlyDictionary<string, SqlTagDefinition> _polledByRawPath =
|
||||
new Dictionary<string, SqlTagDefinition>(StringComparer.Ordinal);
|
||||
|
||||
/// <summary>Resolves a read/subscribe RawPath to its <b>catalog-validated</b> definition, or a miss.</summary>
|
||||
private readonly EquipmentTagRefResolver<SqlTagDefinition> _resolver;
|
||||
|
||||
/// <summary>
|
||||
@@ -182,8 +198,18 @@ public sealed class SqlDriver
|
||||
/// <summary>The current authored-tag snapshot. Read through a barrier — <see cref="BuildTagTable"/> swaps it.</summary>
|
||||
private IReadOnlyDictionary<string, SqlTagDefinition> Tags => Volatile.Read(ref _tagsByRawPath);
|
||||
|
||||
/// <summary>The resolver's lookup: RawPath → authored definition, or null on a miss.</summary>
|
||||
private SqlTagDefinition? Lookup(string rawPath) => Tags.GetValueOrDefault(rawPath);
|
||||
/// <summary>
|
||||
/// The catalog-validated snapshot. Read through a barrier — <see cref="ApplyCatalogGateAsync"/> swaps it.
|
||||
/// </summary>
|
||||
private IReadOnlyDictionary<string, SqlTagDefinition> PolledTags => Volatile.Read(ref _polledByRawPath);
|
||||
|
||||
/// <summary>
|
||||
/// The resolver's lookup: RawPath → <b>catalog-validated</b> definition, or null on a miss.
|
||||
/// <para>Deliberately reads <see cref="PolledTags"/> rather than <see cref="Tags"/>: a tag the gate
|
||||
/// rejected must miss here so the reader publishes
|
||||
/// <see cref="SqlStatusCodes.BadNodeIdUnknown"/>, exactly as design §8.1 specifies.</para>
|
||||
/// </summary>
|
||||
private SqlTagDefinition? Lookup(string rawPath) => PolledTags.GetValueOrDefault(rawPath);
|
||||
|
||||
// ---- IDriver lifecycle ----
|
||||
|
||||
@@ -196,11 +222,16 @@ public sealed class SqlDriver
|
||||
/// I/O, and on failure this method records <see cref="DriverState.Faulted"/> <b>and rethrows</b> —
|
||||
/// <c>DriverInstanceActor</c> reads a throw as InitializeFailed and lands in Reconnecting with its
|
||||
/// retry timer running, which is exactly the recovery a database that is merely down needs.</para>
|
||||
/// <para>The liveness check is then followed by the design §8.1 <b>catalog gate</b>
|
||||
/// (<see cref="ApplyCatalogGateAsync"/>), which is the second and last piece of I/O. It runs after
|
||||
/// liveness because it needs a working connection, and before the driver reports Healthy because the
|
||||
/// tag table it publishes must be the validated one — a poll must never see an unvalidated
|
||||
/// identifier.</para>
|
||||
/// </summary>
|
||||
/// <param name="driverConfigJson">The driver configuration JSON (unused; see remarks).</param>
|
||||
/// <param name="cancellationToken">Cancellation token for the operation.</param>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
/// <exception cref="InvalidOperationException">The database could not be reached.</exception>
|
||||
/// <exception cref="InvalidOperationException">The database could not be reached, or its catalog could not be read.</exception>
|
||||
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
|
||||
{
|
||||
WriteHealth(new DriverHealth(DriverState.Initializing, null, null));
|
||||
@@ -234,6 +265,36 @@ public sealed class SqlDriver
|
||||
throw new InvalidOperationException(message, ex);
|
||||
}
|
||||
|
||||
// Separate try from liveness so the operator surface names the stage that actually failed: "could
|
||||
// not reach the database" and "reached it but could not read its catalog" send an operator to
|
||||
// different places, and the second is usually a GRANT rather than an outage.
|
||||
try
|
||||
{
|
||||
await ApplyCatalogGateAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
WriteHealth(new DriverHealth(DriverState.Unknown, null, null));
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Same discipline as above: exception TYPE only on the operator surface, full exception to the
|
||||
// log sink. A catalog read that fails is NOT evidence the tags are wrong — it is the absence of
|
||||
// evidence — so this faults the driver (and retries) rather than rejecting every tag and serving
|
||||
// a confidently-empty address space.
|
||||
var message =
|
||||
$"Sql driver '{_driverInstanceId}' reached {Endpoint} but could not read its catalog to " +
|
||||
$"validate authored tables/columns: {ex.GetType().Name}. Check that the connecting principal " +
|
||||
"can read the catalog views.";
|
||||
_logger.LogError(ex,
|
||||
"Sql driver {DriverInstanceId} catalog validation against {Endpoint} failed.",
|
||||
_driverInstanceId, Endpoint);
|
||||
WriteHealth(new DriverHealth(DriverState.Faulted, null, message));
|
||||
TransitionHostTo(HostState.Stopped);
|
||||
throw new InvalidOperationException(message, ex);
|
||||
}
|
||||
|
||||
WriteHealth(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
|
||||
TransitionHostTo(HostState.Running);
|
||||
_logger.LogInformation(
|
||||
@@ -594,6 +655,94 @@ public sealed class SqlDriver
|
||||
}
|
||||
|
||||
Volatile.Write(ref _tagsByRawPath, table);
|
||||
|
||||
// Nothing is pollable until the catalog gate has validated it. Clearing here (rather than leaving a
|
||||
// previous generation's validated table in place) means a Reinitialize whose gate then fails cannot
|
||||
// keep polling the OLD identifiers against a database whose schema may be exactly what changed.
|
||||
Volatile.Write(ref _polledByRawPath, new Dictionary<string, SqlTagDefinition>(StringComparer.Ordinal));
|
||||
}
|
||||
|
||||
// ---- catalog gate (design §8.1) ----
|
||||
|
||||
/// <summary>
|
||||
/// Validates every authored identifier against the live catalog and republishes the tag table with
|
||||
/// catalog spellings, dropping the tags that do not resolve (design §8.1).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para><b>Why this must run before the driver reports Healthy.</b> The table this swaps in is what
|
||||
/// every poll resolves against. Running it later — or in the background — would leave a window in
|
||||
/// which an unvalidated identifier could be quoted into a query, which is the entire hole the gate
|
||||
/// exists to close.</para>
|
||||
/// <para><b>A dropped tag is not a driver fault.</b> It disappears from the table, so its RawPath
|
||||
/// resolves to nothing and it publishes <see cref="SqlStatusCodes.BadNodeIdUnknown"/> — one operator
|
||||
/// typo must not stop the other tags on that database, the same rule
|
||||
/// <see cref="BuildTagTable"/> follows for a malformed blob. Each drop is logged at Warning with the
|
||||
/// tag, the field and the reason, because a node that silently goes Bad with nothing in the log is
|
||||
/// unsupportable.</para>
|
||||
/// <para><b>No tags ⇒ no I/O.</b> A driver with nothing authored has nothing to validate, and issuing
|
||||
/// catalog queries to prove that would be a round-trip that can only fail.</para>
|
||||
/// <para>Bounded by the same wall-clock discipline as the liveness check (R2-01 / STAB-14): the work
|
||||
/// runs on the thread pool so a provider implementing the async path synchronously blocks a pool
|
||||
/// thread rather than wedging Initialize forever with no retry.</para>
|
||||
/// </remarks>
|
||||
/// <param name="cancellationToken">The caller's token.</param>
|
||||
/// <returns>A task that completes when the tag table has been validated and republished.</returns>
|
||||
private async Task ApplyCatalogGateAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
var authored = Tags;
|
||||
if (authored.Count == 0) return;
|
||||
|
||||
var tables = authored.Values
|
||||
.Select(definition => definition.Table)
|
||||
.Where(table => !string.IsNullOrWhiteSpace(table))
|
||||
.Distinct(StringComparer.OrdinalIgnoreCase)
|
||||
.ToArray();
|
||||
|
||||
var catalog = await RunBoundedAsync(
|
||||
token => LoadCatalogAsync(tables, token),
|
||||
_options.OperationTimeout,
|
||||
"the catalog queries did not return",
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
|
||||
var result = SqlCatalogGate.Apply(authored.Values, catalog, _dialect);
|
||||
|
||||
foreach (var rejection in result.Rejected)
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"Sql tag '{Tag}' rejected by the catalog gate: {Field} — {Reason}. It will publish "
|
||||
+ "BadNodeIdUnknown until the tag or the database is corrected. Driver={DriverInstanceId}",
|
||||
rejection.RawPath, rejection.Field, rejection.Reason, _driverInstanceId);
|
||||
}
|
||||
|
||||
var validated = new Dictionary<string, SqlTagDefinition>(result.Accepted.Count, StringComparer.Ordinal);
|
||||
foreach (var definition in result.Accepted) validated[definition.Name] = definition;
|
||||
|
||||
// Only the POLLED table is replaced. The authored table is left intact so every authored tag keeps
|
||||
// its node and a rejected one publishes BadNodeIdUnknown rather than vanishing (see _polledByRawPath).
|
||||
Volatile.Write(ref _polledByRawPath, validated);
|
||||
|
||||
_logger.LogInformation(
|
||||
"Sql driver {DriverInstanceId} catalog gate accepted {Accepted} of {Total} authored tag(s) "
|
||||
+ "across {Tables} table(s) on {Endpoint}.",
|
||||
_driverInstanceId, result.Accepted.Count, authored.Count, tables.Length, Endpoint);
|
||||
}
|
||||
|
||||
/// <summary>Opens one connection and reads the catalog slice the authored tables need.</summary>
|
||||
private async Task<SqlCatalog> LoadCatalogAsync(
|
||||
IReadOnlyCollection<string> authoredTables, CancellationToken cancellationToken)
|
||||
{
|
||||
var connection = _factory.CreateConnection()
|
||||
?? throw new InvalidOperationException(
|
||||
$"the {_dialect.Provider} provider factory returned no connection.");
|
||||
|
||||
await using (connection.ConfigureAwait(false))
|
||||
{
|
||||
connection.ConnectionString = _connectionString;
|
||||
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
|
||||
return await SqlCatalogLoader
|
||||
.LoadAsync(connection, _dialect, authoredTables, _options.CommandTimeout, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
// ---- liveness ----
|
||||
@@ -611,18 +760,50 @@ public sealed class SqlDriver
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">The caller's token.</param>
|
||||
/// <returns>A task that completes when the database has answered.</returns>
|
||||
private async Task VerifyLivenessAsync(CancellationToken cancellationToken)
|
||||
private Task VerifyLivenessAsync(CancellationToken cancellationToken) =>
|
||||
RunBoundedAsync(
|
||||
async token => { await PingAsync(token).ConfigureAwait(false); return true; },
|
||||
_options.CommandTimeout,
|
||||
"the liveness statement did not return",
|
||||
cancellationToken);
|
||||
|
||||
/// <summary>
|
||||
/// Runs <paramref name="work"/> under a hard wall-clock <paramref name="budget"/>, on the thread pool,
|
||||
/// converting a breach into a <see cref="TimeoutException"/> worded from <paramref name="what"/>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>The R2-01 / STAB-14 shape, shared by the two pieces of I/O Initialize performs. It exists
|
||||
/// because a token is not a deadline: some ADO.NET providers implement the async path synchronously,
|
||||
/// and a wedged socket can hang <em>inside</em> the provider's own cancellation handshake. Without an
|
||||
/// independent wall clock, <c>DriverInstanceActor</c>'s init task would never complete and the driver
|
||||
/// would sit in Connecting forever with no retry — the exact frozen-peer wedge the S7 driver shipped.</para>
|
||||
/// <para>Both the linked CTS deadline and an outer <see cref="TaskExtensions"/> wait are used, because
|
||||
/// each covers a case the other does not: the CTS handles a provider that <em>does</em> honour
|
||||
/// cancellation, and the outer wait handles one that does not. An abandoned attempt keeps ownership of
|
||||
/// its own connection and disposes it; <see cref="Detach"/> observes its eventual fault.</para>
|
||||
/// </remarks>
|
||||
/// <typeparam name="T">The work's result type.</typeparam>
|
||||
/// <param name="work">The operation to bound. Receives the deadline-linked token.</param>
|
||||
/// <param name="budget">The wall-clock allowance.</param>
|
||||
/// <param name="what">Sentence fragment naming the operation, e.g. "the catalog queries did not return".</param>
|
||||
/// <param name="cancellationToken">The caller's token.</param>
|
||||
/// <returns>The work's result.</returns>
|
||||
/// <exception cref="TimeoutException">The budget elapsed first.</exception>
|
||||
private static async Task<T> RunBoundedAsync<T>(
|
||||
Func<CancellationToken, Task<T>> work,
|
||||
TimeSpan budget,
|
||||
string what,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var budget = _options.CommandTimeout;
|
||||
var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
Task work;
|
||||
Task<T> running;
|
||||
try
|
||||
{
|
||||
deadline.CancelAfter(budget);
|
||||
work = Task.Run(
|
||||
running = Task.Run(
|
||||
async () =>
|
||||
{
|
||||
try { await PingAsync(deadline.Token).ConfigureAwait(false); }
|
||||
try { return await work(deadline.Token).ConfigureAwait(false); }
|
||||
finally { deadline.Dispose(); }
|
||||
},
|
||||
CancellationToken.None);
|
||||
@@ -636,20 +817,18 @@ public sealed class SqlDriver
|
||||
|
||||
try
|
||||
{
|
||||
await work.WaitAsync(budget, cancellationToken).ConfigureAwait(false);
|
||||
return await running.WaitAsync(budget, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (TimeoutException)
|
||||
{
|
||||
Detach(work);
|
||||
throw new TimeoutException(
|
||||
$"the liveness statement did not return within {(int)budget.TotalMilliseconds} ms.");
|
||||
Detach(running);
|
||||
throw new TimeoutException($"{what} within {(int)budget.TotalMilliseconds} ms.");
|
||||
}
|
||||
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
// Our deadline fired and the provider DID honour the linked token.
|
||||
Detach(work);
|
||||
throw new TimeoutException(
|
||||
$"the liveness statement did not return within {(int)budget.TotalMilliseconds} ms.");
|
||||
Detach(running);
|
||||
throw new TimeoutException($"{what} within {(int)budget.TotalMilliseconds} ms.");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -752,6 +931,7 @@ public sealed class SqlDriver
|
||||
{
|
||||
await _poll.DisposeAsync().ConfigureAwait(false);
|
||||
Volatile.Write(ref _tagsByRawPath, new Dictionary<string, SqlTagDefinition>(StringComparer.Ordinal));
|
||||
Volatile.Write(ref _polledByRawPath, new Dictionary<string, SqlTagDefinition>(StringComparer.Ordinal));
|
||||
_resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry
|
||||
}
|
||||
}
|
||||
|
||||
@@ -44,6 +44,12 @@ public sealed class SqlServerDialect : ISqlDialect
|
||||
public string ListSchemasSql =>
|
||||
"SELECT DISTINCT TABLE_SCHEMA FROM INFORMATION_SCHEMA.TABLES ORDER BY TABLE_SCHEMA";
|
||||
|
||||
/// <summary>
|
||||
/// <c>SCHEMA_NAME()</c> with no argument returns the calling principal's default schema — the one an
|
||||
/// unqualified object name resolves in, which is precisely the question the catalog gate asks.
|
||||
/// </summary>
|
||||
public string DefaultSchemaSql => "SELECT SCHEMA_NAME()";
|
||||
|
||||
/// <inheritdoc/>
|
||||
public string ListTablesSql =>
|
||||
"SELECT TABLE_NAME, TABLE_TYPE FROM INFORMATION_SCHEMA.TABLES " +
|
||||
@@ -70,9 +76,11 @@ public sealed class SqlServerDialect : ISqlDialect
|
||||
/// control-character rule.</para>
|
||||
/// <para>The rejection messages deliberately do <b>not</b> echo the offending value — it is untrusted
|
||||
/// input and this exception's text reaches the driver log.</para>
|
||||
/// <para><b>This is currently the only defence, not a backstop.</b> Design §8.1 specifies that an
|
||||
/// identifier is validated against the live catalog before reaching here; that gate is not implemented
|
||||
/// yet, so an authored <c>TagConfig</c> table/column arrives unfiltered.</para>
|
||||
/// <para><b>This is a backstop, not the only defence.</b> Design §8.1's catalog gate
|
||||
/// (<see cref="SqlCatalogGate"/>) resolves an authored table/column against the live catalog at
|
||||
/// Initialize and substitutes the catalog's own spelling, so on the poll path the value arriving here
|
||||
/// is a string read back out of the catalog. The rules below still run — the gate uses them as its own
|
||||
/// charset filter, and they must hold for any future caller that has no catalog to check against.</para>
|
||||
/// </remarks>
|
||||
/// <param name="ident">The bare, unquoted identifier part.</param>
|
||||
/// <returns>The bracket-quoted identifier.</returns>
|
||||
|
||||
Reference in New Issue
Block a user