diff --git a/src/MeshWeaver.Documentation/Data/Architecture/SourceSetEstablishment.md b/src/MeshWeaver.Documentation/Data/Architecture/SourceSetEstablishment.md index 4b68e92189..d3b02a1d75 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/SourceSetEstablishment.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/SourceSetEstablishment.md @@ -42,10 +42,63 @@ anything has been compiled. It is the only independent witness available at disc is used in both directions — the same field, read for both polarities, which is what makes the two classifications agree by construction rather than by coincidence. -`NodeTypeBatchBake.DiscoveryUnestablished` is the one predicate. When it answers true the whole -batch is abandoned with `SourceDiscoveryFailedException` and the pod falls back to the -activation-driven sweep, which re-resolves each type individually — slower, correct, and impossible -to mistake for a content verdict. +`NodeTypeBatchBake.DiscoveryUnestablished` is the one predicate. When it answers true for any type +after the GLOBAL pass, that pass is distrusted as a whole (a short read that empties one type may +have shortened others undetectably) — and the batch then re-resolves every pending type from its OWN +anchored queries, as described next. A type the predicate still refuses is left out of the batch and +warmed by activation, which resolves its sources itself — slower, correct, and impossible to mistake +for a content verdict. + +## The unit of "I don't know" is the type, not the batch + +The global pass is three mesh-wide fetches (`nodeType:Code partitions:all`, and the +`namespace:*/Source` / `namespace:*/Test` subtrees). On a partitioned Postgres mesh each is ONE +`UNION ALL` over every partition schema, so **one partition that cannot answer fails every fetch** — +and until this change that failed every pending type with it. + +Measured on memex-cloud (core `bf5ac85526`, 2026-09-24): on every one of eight boots between 07:12Z +and 08:06Z, `nodeType:Code partitions:all` settled at 1586 nodes and ~25 ms later the pod logged +`batched source discovery did not establish the source sets — abandoning the batch and falling back +to the activation-driven sweep for ALL N pending type(s)`, with N between 38 and 80. That cost each +boot 5–14 minutes of per-type activations; on memex, combined with the activation wait deadlock +fixed in #5643, it held a pod unready for hours. + +So when the global pass faults, times out, reaches its ceiling or is contradicted by a type's +record, `ResolveSources` falls back to `ResolvePerQuery`: every DISTINCT source query the pending +types expand to is run on its own — anchored, so pinned to the one partition it names — with a +concurrency of four, and each type is handed the union of its own queries' answers +(`AssemblePerQuery`). A type joins the batch only when every one of its queries answered and the +answer is established; a type left out is ABSENT from the map, and the sweep warms exactly that type +by activation. It is never handed an empty set, which would be #1216's fabricated verdict. + +Two properties keep the fallback honest: + +- **A floor is not an answer.** Every discovery query (global and per-type) refuses a snapshot whose + frame names `SilentProviders`. That field is set when a provider never answered, or answered over + reads it could not complete (`SnapshotIncomplete`, which is how a partitioned provider reports a + partition it dropped). Folding such a frame would hand the compiler a PARTIAL set. +- **One budget, not one per query.** The whole per-type pass shares ONE deadline of + `DiscoveryBudget`, the same budget the global pass has. Any query still unanswered at that + deadline fails as itself, so stalled queries cannot add up to several budgets before the sweep + starts. + +The fault itself was read off the same boot (pod `…-6f9d6848f8-gf5df`): the line logged 13 µs +after the abandonment at 07:34:15.089Z is `Npgsql.PostgresException (0x80004005): 42703: column +n.created_by does not exist` — the `namespace:*/Source` fetch reads the `code` satellite table of +every schema, and at least one schema's `code` table lacks the authorship columns V57 adds. The +same `42703` recurs on that pod every 60 s from some other mesh-wide read. Which schema, and why V57 +did not reach it, was NOT established. + +What this does NOT do: it does not name or repair the partition that could not answer. The global +fetch's fault is logged at Warning with its exception (`BatchBake: the GLOBAL source-discovery pass +did not establish the source sets () …`), and the types withheld are named with the +query and fault that withheld them — the partition itself is a separate defect in the backend that +answered. + +Pinned by `OnePartitionCostsOnlyItsOwnTypesTest`: a query provider that refuses every mesh-wide +`nodeType:Code` fetch and the anchored reads of ONE type leaves the healthy type in the batch with +its real source set and the broken one absent. With the fallback replaced by a rethrow, the same +test fails with the provider's fault — the pre-fix behaviour. ## What went wrong (issue #3663) diff --git a/src/MeshWeaver.Hosting/DynamicTypePreWarmer.cs b/src/MeshWeaver.Hosting/DynamicTypePreWarmer.cs index 2621487836..2dd30fe0c2 100644 --- a/src/MeshWeaver.Hosting/DynamicTypePreWarmer.cs +++ b/src/MeshWeaver.Hosting/DynamicTypePreWarmer.cs @@ -1223,19 +1223,27 @@ IObservable Sweep( // #5544 — the readiness message names THIS type and the bound it may take, // so a sweep that is slow reads as slow, on the right type, rather than as // "enumerating" for hours. + // + // 🚨 A type ABSENT from batchSources had no source set the discovery could + // establish (its own query failed, or its record contradicts an empty answer). + // It is warmed by activation — ONE type leaves the batch, never all of them, and + // it is never compiled against an empty set (that would be #1216's fabricated + // verdict). + IReadOnlyList? batchSet = null; + MeshNode? typeNode = null; + var inBatch = batchSources is not null + && batchSources.TryGetValue(p, out batchSet) + && nodes.TryGetValue(p, out typeNode); progress?.Invoke( $"building {pending.IndexOf(p) + 1} of {pending.Count} pending NodeType(s) " - + (batchSources is not null + + (inBatch ? "by direct batch compile" : bytesMissing.Contains(p) ? "by a store-miss rebuild" - : "by activation (the batch was not available)") + : "by activation (the batch did not establish its sources)") + $": waiting on {p}, for up to {budget}"); - var warm = batchSources is not null && nodes.TryGetValue(p, out var typeNode) - ? NodeTypeBatchBake.BakeOne( - mesh, typeNode, - batchSources.TryGetValue(p, out var srcs) ? srcs : Array.Empty(), - budget, logger) + var warm = inBatch + ? NodeTypeBatchBake.BakeOne(mesh, typeNode!, batchSet!, budget, logger) : missingBytes ? RebuildMissingBytes(workspace, accessService, p, budget, logger) : WarmOne(workspace, accessService, p, budget, logger); @@ -1243,6 +1251,8 @@ IObservable Sweep( return warm .DelaySubscription( i == 0 || batchSources is not null ? TimeSpan.Zero : pacing) + // (A type the batch withheld is activated WITHOUT pacing too: batchSources is + // non-null only on a gated pod, which serves nobody while it bakes.) // 🚨 A verdict against a node that NO LONGER EXISTS is not a verdict about // the image. The definitions were enumerated once, at the start of the // sweep; a repository sync can prune a retired type at any moment after @@ -1288,8 +1298,13 @@ or PreWarmStatus.Retired // does not know what the sources ARE, and a compile driven from a set you did not establish // produces verdicts about code from evidence you do not have — on memex-cloud 2026-08-11 // that read as "169 of 237 types are content-broken" and, on an ungated pod, would have - // baked a fleet of empty assemblies with nothing refusing readiness. Whole-batch fallback is - // the only safe answer: the activation-driven sweep resolves each type's sources itself. + // baked a fleet of empty assemblies with nothing refusing readiness. The unit of that "I don't + // know" is the TYPE, not the batch: ResolveSources re-resolves every type from its own + // anchored queries when the global pass cannot be trusted, and leaves out of the map only the + // types it still could not establish — those, and only those, take the activation-driven + // sweep (which resolves their sources itself). Measured on memex-cloud 2026-09-24: one + // unanswerable global fetch sent ALL 38–80 pending types of every boot down that road. What + // still reaches the Catch below is a fault in discovery's own composition, not a query's. IObservable>?> BatchSources() => Observable.Defer(() => { diff --git a/src/MeshWeaver.Hosting/NodeTypeBatchBake.cs b/src/MeshWeaver.Hosting/NodeTypeBatchBake.cs index ce9b8ae02c..6a85e18429 100644 --- a/src/MeshWeaver.Hosting/NodeTypeBatchBake.cs +++ b/src/MeshWeaver.Hosting/NodeTypeBatchBake.cs @@ -170,7 +170,7 @@ internal static class NodeTypeBatchBake /// change — the in-memory match handles only the shapes it provably mirrors /// (path:/namespace:/scope:/nodeType:Code). /// - /// Per-type source lists keyed by type path; missing key = no sources resolved. + /// Per-type source lists keyed by type path. A pending type ABSENT from the map had no source set this pass could establish, and must be warmed by activation — never compiled against nothing. public static IObservable>> ResolveSources( IMeshService meshService, AccessService? accessService, @@ -199,7 +199,7 @@ public static IObservable>> /// The types this batch is discovering sources for. /// Logger, or null. /// Where each pass's timing is published, or null to publish none. - /// Per-type source lists keyed by type path; missing key = no sources resolved. + /// Per-type source lists keyed by type path. A pending type ABSENT from the map had no source set this pass could establish, and must be warmed by activation — never compiled against nothing. public static IObservable>> ResolveSources( IMeshService meshService, AccessService? accessService, @@ -285,12 +285,167 @@ public static IObservable>> .Select(results => results.ToImmutableDictionary( r => r.Query, r => r.Nodes, StringComparer.Ordinal)); + // 🚨 THE GLOBAL PASS IS AN OPTIMISATION, NEVER A PRECONDITION. Its three fetches + // are mesh-wide by construction — on a partitioned Postgres mesh each one is ONE + // UNION ALL over every partition schema — so ONE partition that cannot answer + // fails ALL of them, and with them every pending type, including every type whose + // sources live in partitions that answer perfectly well. Measured on memex-cloud + // (core bf5ac85526, 2026-09-24): on every one of eight boots between 07:12Z and + // 08:06Z `nodeType:Code partitions:all` settled at 1586 nodes and the pass was + // abandoned ~25 ms later, before the `namespace:*/Source` fetch settled — and + // each abandonment cost the boot the activation-driven sweep for ALL of its 38–80 + // pending types (5–14 minutes per boot; on memex, with #5643's wait deadlock, + // hours). + // + // So a global pass that cannot be TRUSTED — it faulted, it timed out, it hit its + // ceiling, or one type's record contradicts it (#3663: then the read is short for + // reasons nobody can see, and a PARTIAL set elsewhere is undetectable) — is not + // the end of the batch. Every pending type is re-resolved from its OWN anchored + // queries (ResolvePerQuery): the queries the compiler runs for it, each pinned to + // the partition it names, so a partition that cannot answer takes down exactly + // the types that read from it. Those — and only those — leave the batch for the + // activation-driven sweep; the rest are still baked directly. return globalFetch .SelectMany(allCode => exoticFetch.Select(exotic => Assemble(perType, allCode, exotic, logger))) - .Timeout(DiscoveryBudget); + .Timeout(DiscoveryBudget) + .Catch>, Exception>(ex => + { + logger?.LogWarning(ex, + "BatchBake: the GLOBAL source-discovery pass did not establish the " + + "source sets ({Reason}) — re-resolving each of the {Count} pending " + + "type(s) from its OWN anchored source queries, so only the types " + + "whose own queries cannot be answered leave the batch", + ex.GetType().Name, perType.Count); + return ResolvePerQuery(meshService, perType, logger, discovery); + }); }); } + /// How many of the per-type anchored discovery queries run at once. Each one is pinned + /// to ONE partition and spends most of its life in , so running + /// them strictly one after another would cost about one second per query for nothing. + private const int PerQueryConcurrency = 4; + + /// + /// The per-type discovery pass — what the batch does when the global pass cannot be trusted. + /// Runs every DISTINCT source query the pending types expand to, each against the mesh on its + /// own (anchored, so pinned to the partition it names), and hands each type the union of ITS + /// queries' answers. A query that faults, times out or hits its ceiling is recorded as a + /// failure of THAT query; then withholds exactly the types that + /// depend on it. + /// + private static IObservable>> ResolvePerQuery( + IMeshService meshService, + IReadOnlyList perType, + ILogger? logger, + SourceDiscoveryRegistry? discovery) + { + var queries = perType + .SelectMany(t => t.Matchable.Concat(t.Exotic)) + .Distinct(StringComparer.Ordinal) + .ToList(); + if (queries.Count == 0) + return Observable.Return(AssemblePerQuery( + perType, ImmutableDictionary.Empty, logger)); + + // ONE deadline for the whole per-type pass, shared by every query — the same budget the + // global pass has, never one per query: a set of stalled queries must not add up to + // ceil(queries / concurrency) budgets before the sweep starts. A query still unanswered at + // the deadline (running or not yet started) fails as ITSELF, so its types alone are withheld. + return Observable.Defer(() => + { + var deadline = DateTimeOffset.UtcNow + DiscoveryBudget; + return PerQueryAnswers(meshService, queries, deadline, logger, discovery); + }) + .Select(answers => AssemblePerQuery(perType, answers, logger)); + } + + private static IObservable> PerQueryAnswers( + IMeshService meshService, + IReadOnlyList queries, + DateTimeOffset deadline, + ILogger? logger, + SourceDiscoveryRegistry? discovery) + => queries + .Select(q => RunQuery(meshService, q, logger, discovery) + .Timeout(deadline) + .Select(nodes => new QueryAnswer(nodes, null)) + // Not a swallow: the fault becomes THIS query's recorded answer, and every type that + // reads from it is withheld from the batch and named, with this fault, below. + .Catch(ex => Observable.Return(new QueryAnswer(null, ex))) + .Select(answer => (Query: q, Answer: answer))) + .MergeBounded(PerQueryConcurrency) + .ToList() + .Select(answers => answers.ToImmutableDictionary( + a => a.Query, a => a.Answer, StringComparer.Ordinal)); + + /// One discovery query's outcome: its settled node map, or the fault that stopped it. + /// The settled path→node map, or null when the query failed. + /// Why the query produced no map, or null when it settled. + internal sealed record QueryAnswer(ImmutableDictionary? Nodes, Exception? Fault); + + /// + /// 🚨 Per-type assembly: a type joins the batch only when EVERY one of its own queries + /// answered and the answer is established (). A type left + /// out is NOT a verdict — the sweep warms it by activation, exactly as the whole batch used to be + /// warmed when one partition failed. Pure, so the degradation rule is pinned without a mesh. + /// + /// The pending types and their expanded queries. + /// Each distinct query's outcome, keyed by the query text. + /// Logger, or null. + /// The established source sets; a pending type ABSENT from the map was not established. + internal static ImmutableDictionary> AssemblePerQuery( + IReadOnlyList perType, + IReadOnlyDictionary answers, + ILogger? logger) + { + var builder = ImmutableDictionary.CreateBuilder>( + StringComparer.OrdinalIgnoreCase); + var withheld = ImmutableList.Empty; + foreach (var pending in perType) + { + var matched = ImmutableDictionary.Empty + .WithComparers(StringComparer.OrdinalIgnoreCase); + string? failure = null; + foreach (var q in pending.Matchable.Concat(pending.Exotic)) + { + if (!answers.TryGetValue(q, out var answer) || answer.Nodes is null) + { + failure = $"its source query '{q}' could not be answered" + + (answer?.Fault is { } fault ? $" ({fault.GetType().Name}: {fault.Message})" : ""); + break; + } + matched = matched.SetItems(answer.Nodes); + } + + if (failure is null + && DiscoveryUnestablished(matched.Count, pending.DeclaresSources, pending.KnownSourceCount)) + failure = "its own queries resolved an EMPTY set that its record contradicts"; + + if (failure is not null) + { + withheld = withheld.Add($"{pending.TypePath} — {failure}"); + continue; + } + + builder[pending.TypePath] = matched.Values + .OrderBy(n => n.Path, StringComparer.Ordinal) + .ToImmutableList(); + } + + logger?.LogInformation( + "BatchBake: per-type source discovery established {Resolved} of {Total} pending type(s) " + + "from {Queries} anchored queries", + builder.Count, perType.Count, answers.Count); + if (!withheld.IsEmpty) + logger?.LogWarning( + "BatchBake: {Count} pending type(s) are warmed by ACTIVATION instead of the batch, " + + "because their source set could not be established — not a content or compile " + + "verdict: {Types}", + withheld.Count, string.Join("; ", withheld)); + return builder.ToImmutable(); + } + /// /// A query the in-memory matcher provably mirrors: every token is path:, /// namespace:, scope: or nodeType:Code. Anything else (free text, other @@ -354,7 +509,7 @@ internal static bool IsInMemoryMatchable(string expandedQuery) /// corroboration that "no matches" is the mesh's real answer, a positive value is independent /// evidence that it is NOT, and null (absent snapshot) is no evidence either way. /// - private sealed record PendingType( + internal sealed record PendingType( string TypePath, IReadOnlyList Matchable, IReadOnlyList Exotic, @@ -409,6 +564,7 @@ private static IObservable> RunQuery( // concurrent discoveries of the same query text must not fold their chunk timings // together, and nothing here is static (#3704). var chunks = new ChunkTiming(); + var silent = ImmutableHashSet.Empty.WithComparer(StringComparer.Ordinal); return Observable // 🚨 .AsSystem() — batch-bake source discovery is framework infrastructure, not a // user-scoped read, and this Defer's subscription does not carry the caller's ambient @@ -422,13 +578,30 @@ private static IObservable> RunQuery( Limit = SourceDiscoveryLimit, }.AsSystem())) .Do(chunks.Observe) + // 🚨 A FLOOR IS NOT A SOURCE SET. A frame naming SilentProviders was answered over + // reads that did not complete — a provider that never answered, or one that caught a + // partition's read fault, dropped those rows and emitted what survived + // (SnapshotIncomplete, folded into SilentProviders by the fan-in). Folded as an + // answer it hands the compiler a PARTIAL set, which compiles wrong in the most + // convincing way (#1218). Recorded per subscription, then refused below. + .Do(change => + { + if (change.SilentProviders is { Count: > 0 } s) + silent = silent.Union(s); + }) .Scan( ImmutableDictionary.Empty.WithComparers(StringComparer.OrdinalIgnoreCase), NodeCompileShaping.ApplyQueryChange) .Throttle(QueryQuietWindow) .Take(1) .Do(nodes => chunks.Report(logger, bounded, nodes.Count, QueryQuietWindow, discovery)) - .SelectMany(nodes => nodes.Count < SourceDiscoveryLimit + .SelectMany(nodes => !silent.IsEmpty + ? Observable.Throw>( + new SourceDiscoveryFailedException( + $"source discovery query '{bounded}' settled at {nodes.Count} node(s) over a " + + $"snapshot provider(s) [{string.Join(", ", silent.Order(StringComparer.Ordinal))}] " + + "did not complete, so it is a FLOOR, not a source set")) + : nodes.Count < SourceDiscoveryLimit ? Observable.Return(nodes) : Observable.Throw>( new SourceDiscoveryFailedException( diff --git a/test/MeshWeaver.Hosting.Test/OnePartitionCostsOnlyItsOwnTypesTest.cs b/test/MeshWeaver.Hosting.Test/OnePartitionCostsOnlyItsOwnTypesTest.cs new file mode 100644 index 0000000000..07194131e4 --- /dev/null +++ b/test/MeshWeaver.Hosting.Test/OnePartitionCostsOnlyItsOwnTypesTest.cs @@ -0,0 +1,222 @@ +using System; +using System.Collections.Generic; +using System.Collections.Immutable; +using System.Linq; +using System.Reactive.Linq; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using MeshWeaver.Data; +using MeshWeaver.Fixture; +using MeshWeaver.Graph.Configuration; +using MeshWeaver.Hosting.Monolith.TestBase; +using MeshWeaver.Mesh; +using MeshWeaver.Mesh.Services; +using MeshWeaver.Messaging; +using MeshWeaver.Reactive.Assertions; +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace MeshWeaver.Hosting.Test; + +/// +/// 🚨 One partition that cannot answer costs the batched bake ITS OWN types — never all of +/// them. +/// +/// Measured on memex-cloud, core bf5ac85526, 2026-09-24: on every one of eight +/// boots between 07:12Z and 08:06Z the batched source discovery's first global fetch +/// (nodeType:Code partitions:all) settled at 1586 nodes and the pass was abandoned ~25 ms +/// later — DynamicTypePreWarmer: batched source discovery did not establish the source sets — +/// abandoning the batch and falling back to the activation-driven sweep for ALL 53 pending +/// type(s). Each global fetch is ONE UNION over every partition schema, so a single partition +/// whose read fails fails every fetch, and ResolveSources surfaced that as a fault of the +/// whole batch: 38–80 pending types per boot went to the activation-driven sweep, 5–14 minutes a +/// boot, and on memex (with #5643's wait deadlock) hours. +/// +/// What this pins. The provider below is a test PROVIDER (the extension point every +/// backend plugs into — the same seam ColdEmptyInitialIsNotCachedTest drives), not a mock of +/// a core service: it fails exactly the reads a partitioned backend fails when one partition cannot +/// answer — every mesh-wide nodeType:Code fetch, plus the anchored reads of the one broken +/// type. Everything else is the real mesh, the real query fan-in and the sweep's own discovery. The +/// healthy type must come back with its real source set; the broken one must be ABSENT (so the sweep +/// warms it by activation), never present with an empty set (that would be #1216's fabricated +/// verdict). On the pre-fix code the whole call faults, so the first assertion fails — the negative +/// control is the defect itself. +/// +public class OnePartitionCostsOnlyItsOwnTypesTest(ITestOutputHelper output) : MonolithMeshTestBase(output) +{ + private readonly string suffix = Guid.NewGuid().ToString("N")[..8]; + private string GoodType => $"{TestPartition}/Healthy{suffix}"; + private string BrokenType => $"{TestPartition}/Broken{suffix}"; + + private readonly UnanswerablePartitionProvider provider = new(); + + protected override MeshBuilder ConfigureMesh(MeshBuilder builder) => base.ConfigureMesh(builder) + .ConfigureServices(services => services.AddSingleton(provider)); + + private IMeshService MeshService => Mesh.ServiceProvider.GetRequiredService(); + + private static MeshNode TypeNode(string id) => new(id, TestPartition) + { + NodeType = MeshNode.NodeTypePath, + Name = id, + State = MeshNodeState.Active, + Content = new NodeTypeDefinition { Configuration = "config => config" }, + }; + + private static MeshNode Code(string ns) => new("Main", ns) + { + NodeType = "Code", + State = MeshNodeState.Active, + Content = new CodeConfiguration + { + Language = "csharp", + Code = "public static class Answers { public static int Answer() => 42; }", + }, + }; + + [Fact(Timeout = 240_000)] + public async Task AGlobalFetchThatCannotBeAnswered_WithholdsOnlyTheTypesWhoseOwnQueriesFail() + { + var ct = TestContext.Current.CancellationToken; + foreach (var node in new[] + { + TypeNode($"Healthy{suffix}"), Code($"{GoodType}/Source"), + TypeNode($"Broken{suffix}"), Code($"{BrokenType}/Source"), + }) + await MeshService.CreateNode(node).Take(1) + .Should().Within(TestTimeouts.Convergence) + .Emit($"{node.Path} must exist before discovery reads it", ct); + await UntilListed($"{GoodType}/Source/Main", ct); + + // Armed only now, so the seeding above reads the mesh as it is. + provider.Arm(BrokenType); + + var access = Mesh.ServiceProvider.GetRequiredService(); + var definitions = ImmutableDictionary.Empty + .Add(GoodType, new NodeTypeDefinition { Configuration = "config => config" }) + .Add(BrokenType, new NodeTypeDefinition { Configuration = "config => config" }); + var sets = await NodeTypeBatchBake + .ResolveSources(MeshService, access, definitions, [GoodType, BrokenType], null) + .Should().Within(TestTimeouts.Convergence) + .Emit("a partition that cannot answer must not fault the whole discovery pass", ct); + + Output.WriteLine("established: {0}", string.Join(", ", sets.Keys)); + Assert.True(provider.GlobalFetchesRefused > 0, + "the mesh-wide fetch must actually have been refused, or this test measures nothing"); + Assert.True(sets.TryGetValue(GoodType, out var good), + "the healthy type's own anchored query answers, so it stays in the batch"); + Assert.Equal([$"{GoodType}/Source/Main"], good!.Select(n => n.Path).ToArray()); + Assert.False(sets.ContainsKey(BrokenType), + "the type whose own query fails is withheld (warmed by activation), never handed an empty set"); + } + + /// + /// The per-type rule on its own, without a mesh: a type is in the map only when EVERY one of its + /// queries answered and the answer is established; a failed query withholds the types that read + /// it and nobody else, and a contradicted empty answer withholds its type as before (#3663). + /// + [Fact] + public void AssemblePerQuery_WithholdsExactlyTheTypesWhoseOwnQueriesFailed() + { + const string qA = "namespace:P/A/Source scope:subtree nodeType:Code"; + const string qB = "namespace:Q/B/Source scope:subtree nodeType:Code"; + const string qC = "namespace:P/C/Source scope:subtree nodeType:Code"; + var node = new MeshNode("Main", "P/A/Source") { NodeType = "Code" }; + var answers = ImmutableDictionary.Empty + .Add(qA, new(ImmutableDictionary.Empty.Add(node.Path, node), null)) + .Add(qB, new(null, new InvalidOperationException("42703: column n.created_by does not exist"))) + .Add(qC, new(ImmutableDictionary.Empty, null)); + ImmutableList perType = + [ + new("P/A", [qA], [], DeclaresSources: false, KnownSourceCount: 1), + new("Q/B", [qB], [], DeclaresSources: false, KnownSourceCount: 1), + // Empty, and the record says it HAS two files: unestablished, withheld. + new("P/C", [qC], [], DeclaresSources: false, KnownSourceCount: 2), + ]; + + var sets = NodeTypeBatchBake.AssemblePerQuery(perType, answers, null); + + Assert.Equal(["P/A"], sets.Keys.ToArray()); + Assert.Equal([node.Path], sets["P/A"].Select(n => n.Path).ToArray()); + } + + private async Task UntilListed(string path, CancellationToken ct) + => await Observable.Interval(200.Milliseconds()).StartWith(0L) + .SelectMany(_ => MeshService + .Query(MeshQueryRequest.FromQuery($"path:{path}").AsSystem()) + .Take(1)) + .Where(change => change.Items.Any(n => n.Path == path)) + .FirstAsync() + .Should().Within(TestTimeouts.Convergence) + .Emit($"discovery must be able to see {path}", ct); + + /// + /// Answers every query with an empty Initial — except, once armed, the reads a partitioned + /// backend fails when one partition cannot answer: every MESH-WIDE nodeType:Code fetch + /// (one UNION over every partition), and the anchored reads of the one broken type. + /// + private sealed class UnanswerablePartitionProvider : IMeshQueryProvider + { + private string? broken; + private int refused; + + public string Name => nameof(UnanswerablePartitionProvider); + + public int GlobalFetchesRefused => Volatile.Read(ref refused); + + public void Arm(string brokenType) => Volatile.Write(ref broken, brokenType); + + public bool Matches(IReadOnlyList queryNamespaces) => true; + + public IObservable> Query( + MeshQueryRequest request, JsonSerializerOptions options) => + Observable.Defer(() => + { + var brokenType = Volatile.Read(ref broken); + var queries = request.EffectiveQueries.ToList(); + var code = queries.Any(q => q.Contains("nodeType:Code", StringComparison.OrdinalIgnoreCase)); + if (brokenType is not null && code) + { + var meshWide = queries.Any(q => + q.Contains(ParsedQuery.CrossPartitionQualifier, StringComparison.OrdinalIgnoreCase) + || q.Contains("namespace:*", StringComparison.OrdinalIgnoreCase)); + if (meshWide) + Interlocked.Increment(ref refused); + if (meshWide) + return Observable.Throw>(new InvalidOperationException( + "42703: column n.created_by does not exist (one partition cannot answer)")); + // The broken type's OWN read answers the way a partitioned provider answers a + // read it could not complete: an Initial over the rows that survived, marked + // incomplete — the shape that must NOT be folded as a source set. + if (queries.Any(q => q.Contains(brokenType, StringComparison.OrdinalIgnoreCase))) + return Observable.Return(new QueryResultChange + { + ChangeType = QueryChangeType.Initial, + Items = Array.Empty(), + Timestamp = DateTimeOffset.UtcNow, + SnapshotIncomplete = true, + }); + } + return Observable.Return(new QueryResultChange + { + ChangeType = QueryChangeType.Initial, + Items = Array.Empty(), + Timestamp = DateTimeOffset.UtcNow, + }); + }); + + public IObservable> Query( + MeshQueryRequest request, JsonSerializerOptions options) + => Observable.Return((IReadOnlyCollection)Array.Empty()); + + public IObservable> Autocomplete( + string basePath, string prefix, JsonSerializerOptions options, + AutocompleteMode mode = AutocompleteMode.RelevanceFirst, int limit = 10, + string? contextPath = null, string? context = null) + => Observable.Return((IReadOnlyCollection)Array.Empty()); + + public IObservable Select(string path, string property, JsonSerializerOptions options) + => Observable.Return(default); + } +}