Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
149 changes: 109 additions & 40 deletions src/MeshWeaver.Hosting/Persistence/InMemoryStorageAdapter.cs
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,29 @@ public sealed class InMemoryStorageAdapter : SimpleMeshNodeStorage, IStorageAdap
// whenever a write lands during a teardown. Symptom: a LIVE children query that silently stops
// re-emitting after a create that completed successfully.
private readonly IsolatedChangeFeed _changes;
// 🚨 The CHILDREN INDEX (2026-09-13). ListChildPaths used to walk EVERY key of the store on
// every call — a full scan with a Split per key — and the storage-adapter query provider calls
// it once per directory level of every subtree query. The plugin gate (mw-plugin-test) runs on
// this adapter with ~1,100 nodes and issues subtree queries per package install, so that scan
// was quadratic in the mesh size: 12.5 % of the tester's CPU samples in a dotnet-trace of one
// package (Hosting, 198 nodes), and the shard's 18 install-minutes scaled 4× from a laptop to
// a 2-vCPU runner. The index maps a directory to its immediate children (node paths AND implied
// directories) and is maintained by every write path of this adapter. It is keyed by the
// DICTIONARY instance, not the adapter, because several adapters can be built over one shared
// dictionary (the Orleans test cluster's silos): one index per store, every writer maintains
// it. A dictionary mutated behind the adapters' back (a test seeding the map directly) is
// caught by the count check in ChildrenOf, which rebuilds the whole index once.
private static readonly System.Runtime.CompilerServices.ConditionalWeakTable<
ConcurrentDictionary<string, MeshNode>, ChildIndex> Indexes = new();
private readonly ChildIndex _index;

private sealed class ChildIndex
{
public readonly ConcurrentDictionary<string, ConcurrentDictionary<string, byte>> Children =
new(StringComparer.OrdinalIgnoreCase);
public int IndexedCount = -1;
public readonly object Rebuild = new();
}

/// <inheritdoc />
public IObservable<DataChangeNotification> Changes => _changes;
Expand Down Expand Up @@ -66,13 +89,80 @@ public InMemoryStorageAdapter(
_nodes = nodes;
_partitionObjects = partitionObjects;
_logger = logger;
_index = Indexes.GetValue(nodes, _ => new ChildIndex());
// The logger is passed on DELIBERATELY: an isolated fault that nobody logs is the silence
// this feed exists to end (see PostgreSqlPartitionStorageProvider's null-logger regression).
_changes = new IsolatedChangeFeed(logger, "in-memory");
}

private static string Norm(string? path) => path?.Trim('/') ?? "";

/// <summary>Records <paramref name="path"/> under every ancestor directory (the root is "").</summary>
private void Index(string path)
{
var child = path;
while (true)
{
var slash = child.LastIndexOf('/');
var parent = slash < 0 ? "" : child[..slash];
var set = _index.Children.GetOrAdd(parent, _ => new ConcurrentDictionary<string, byte>(StringComparer.OrdinalIgnoreCase));
var added = set.TryAdd(child, 0);
if (slash < 0 || !added)
return; // the root, or an ancestor chain that is already indexed
child = parent;
}
}

/// <summary>Drops <paramref name="path"/> from its parent; an ancestor left with nothing under it is dropped too.</summary>
private void Unindex(string path)
{
var child = path;
while (true)
{
if (_index.Children.TryGetValue(child, out var own) && !own.IsEmpty)
return; // still a directory with descendants — keep the chain
_index.Children.TryRemove(child, out _);
var slash = child.LastIndexOf('/');
var parent = slash < 0 ? "" : child[..slash];
if (_index.Children.TryGetValue(parent, out var set))
set.TryRemove(child, out _);
if (slash < 0 || _nodes.ContainsKey(parent) || (set is not null && !set.IsEmpty))
return;
child = parent;
}
}

/// <summary>The indexed children of a directory, rebuilding the index once if the store was mutated directly.</summary>
private ConcurrentDictionary<string, byte>? ChildrenOf(string parent)
{
if (_index.IndexedCount != _nodes.Count)
{
lock (_index.Rebuild)
{
if (_index.IndexedCount != _nodes.Count)
{
_index.Children.Clear();
foreach (var k in _nodes.Keys)
Index(k);
_index.IndexedCount = _nodes.Count;
}
}
}
return _index.Children.TryGetValue(parent, out var set) ? set : null;
}

private void Added(string path)
{
Index(path);
_index.IndexedCount = _nodes.Count;
}

private void Removed(string path)
{
Unindex(path);
_index.IndexedCount = _nodes.Count;
}

/// <inheritdoc />
public override IObservable<MeshNode?> Read(string path, JsonSerializerOptions options)
=> Observable.Defer(() =>
Expand Down Expand Up @@ -137,6 +227,7 @@ public InMemoryStorageAdapter(
return Observable.Return<MeshNode?>(winner);
}

Added(Norm(node.Path));
_logger?.LogDebug("[InMemoryAdapter#{Id:X}] Write {Path} (count={Count})",
GetHashCode(), Norm(node.Path), _nodes.Count);
// No try/catch: IsolatedChangeFeed already isolates and LOGS a faulty observer, so a
Expand Down Expand Up @@ -181,6 +272,7 @@ public InMemoryStorageAdapter(
GetHashCode(), path);
return Observable.Return<bool?>(false);
}
Added(path);
_changes.OnNext(DataChangeNotification.Updated(path, node));
return Observable.Return<bool?>(true);
}
Expand Down Expand Up @@ -235,7 +327,8 @@ private static MeshNode MaterializeContent(MeshNode node, JsonSerializerOptions
public override IObservable<string> Delete(string path)
=> Observable.Defer(() =>
{
_nodes.TryRemove(Norm(path), out var removed);
if (_nodes.TryRemove(Norm(path), out var removed))
Removed(Norm(path));
_changes.OnNext(DataChangeNotification.Deleted(Norm(path), removed));
return Observable.Return(path);
});
Expand All @@ -251,7 +344,10 @@ public IObservable<bool> DeleteIfExists(string path)
{
var won = _nodes.TryRemove(Norm(path), out var removed);
if (won)
{
Removed(Norm(path));
_changes.OnNext(DataChangeNotification.Deleted(Norm(path), removed));
}
return Observable.Return(won);
});

Expand All @@ -261,53 +357,26 @@ public IObservable<bool> DeleteIfExists(string path)
=> Observable.Defer(() =>
{
var normalized = Norm(parentPath);
var prefix = string.IsNullOrEmpty(normalized) ? "" : normalized + "/";
var expectedDepth = string.IsNullOrEmpty(normalized)
? 1
: normalized.Split('/', StringSplitOptions.RemoveEmptyEntries).Length + 1;

var nodePaths = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
// 🚨 DirectoryPaths must include any intermediate prefix that has at
// least one descendant node — a stored node at depth N≥expectedDepth+1
// implies a "directory" at the expectedDepth level even if no node
// lives there (e.g. SaveNode("org/acme/project/web") doesn't store
// "org/acme/project" but WalkDescendants must recurse into it to find
// "web"/"mobile"). Without this, GetDescendants returns empty for
// any tree whose structure has "directory" levels.
// 🚨 DirectoryPaths must include any intermediate prefix that has at least one
// descendant node — a stored node at depth N≥expectedDepth+1 implies a "directory" at
// the expectedDepth level even if no node lives there (SaveNode("org/acme/project/web")
// stores no "org/acme/project", yet WalkDescendants must recurse into it). The index
// carries exactly those implied directories: a child that is not itself a node.
var directoryPaths = new HashSet<string>(StringComparer.OrdinalIgnoreCase);

foreach (var k in _nodes.Keys)
{
if (!string.IsNullOrEmpty(prefix) && !k.StartsWith(prefix, StringComparison.OrdinalIgnoreCase))
continue;
if (string.IsNullOrEmpty(prefix) && k.Contains('/'))
{
// root level: path with '/' — top segment is a directory
directoryPaths.Add(k.Split('/', 2)[0]);
continue;
}
var segments = k.Split('/', StringSplitOptions.RemoveEmptyEntries);
if (segments.Length == expectedDepth)
nodePaths.Add(k);
else if (segments.Length > expectedDepth)
var children = ChildrenOf(normalized);
if (children is not null)
foreach (var child in children.Keys)
{
// intermediate segment at expectedDepth becomes a directory entry
var dirPath = string.Join("/", segments.Take(expectedDepth));
if (!_nodes.ContainsKey(dirPath))
directoryPaths.Add(dirPath);
if (_nodes.ContainsKey(child))
nodePaths.Add(child);
else if (_index.Children.TryGetValue(child, out var own) && !own.IsEmpty)
directoryPaths.Add(child);
}
}

return Observable.Return<(IEnumerable<string>, IEnumerable<string>)>(
(nodePaths, directoryPaths));
});

/// <summary>
/// Native descendant enumeration: a direct prefix scan over the path-keyed
/// store — exact and race-free against the dictionary that IS the storage of
/// record. Declared on this class (not inherited from the base) for the same
/// interface-slot reason as <see cref="ResolvePath"/> below.
/// </summary>
public IObservable<IReadOnlyCollection<string>> ListDescendantPaths(string rootPath)
=> Observable.Defer(() =>
{
Expand Down
8 changes: 7 additions & 1 deletion src/MeshWeaver.Messaging.Hub/MessageHub.cs
Original file line number Diff line number Diff line change
Expand Up @@ -938,7 +938,13 @@ public bool FailGate(string name, string reason)
/// LogLevel.Trace flooding. Tuned so chat / layout / routing hops only log
/// when something is genuinely slow.
/// </summary>
private static readonly long SlowDispatchTicks = (long)(TimeSpan.TicksPerMillisecond * 500);
// 🚨 In STOPWATCH ticks, because it is compared against `Stopwatch.GetTimestamp()` deltas
// (below). It used to be `TimeSpan.TicksPerMillisecond * 500` — 100-ns ticks — which on Linux
// (Stopwatch.Frequency = 1e9) is 5 ms, so every dispatch over 5 ms logged as SLOW: 366 lines
// per package install, 16,000 lines per gate shard, all at Information (measured
// 2026-09-13 on Plugins run 34731952463). The threshold the comment above describes — 500 ms
// — is what this now is, on every platform.
private static readonly long SlowDispatchTicks = Stopwatch.Frequency / 2;

// Reactive end-to-end: IObservable, no async/await, no Task in the signature.
// Runs INLINE on the turn thread (Defer → factory on Subscribe); a synchronous
Expand Down
16 changes: 15 additions & 1 deletion src/MeshWeaver.PluginCatalog/PackageInstaller.cs
Original file line number Diff line number Diff line change
Expand Up @@ -2765,7 +2765,21 @@ int Order(MeshNode n)
var rootTypeIsStatic = root is null
|| string.IsNullOrEmpty(root.NodeType)
|| hub.ServiceProvider.FindStaticNode(root.NodeType!) is not null;
var placeholderRoot = root is not null && !rootTypeIsStatic
// 🚨 The placeholder dance exists for ONE case: a root whose NodeType is defined IN THIS
// package, so the type cannot be loadable before the package's own NodeType node lands and
// compiles — the root is written as a Space first, retyped in stage 2, and its hub recycled
// (SettleRetypedRoot). A root typed by ANOTHER package (every satellite of Store: Hosting,
// Signature, Essentials … are Store/Plugin) needs none of it — the type is registered or
// it is not, exactly as for any instance — yet until 2026-09-13 every non-static root took
// the dance: one placeholder write, one retype, one hub recycle per package, and behind the
// recycle a batch of [UpdateQueue] ADVANCE_WITHOUT_HANDOFF bound=5000ms waits on the
// compile-state writes of the package's types (13 s per package on the plugin gate,
// measured on Plugins run 34731952463). Maintainer: "must be done on running mesh not ramp
// up and down". So: a placeholder only when the root's type is in-package.
var rootTypeInPackage = root is not null && !string.IsNullOrEmpty(root.NodeType)
&& nodes.Any(n => ImportWriteOrder.IsNodeTypeDefinition(n)
&& string.Equals(n.Path, root.NodeType, StringComparison.Ordinal));
var placeholderRoot = root is not null && !rootTypeIsStatic && rootTypeInPackage
? root with { NodeType = "Space", Content = null }
: null;
var stage0 = root is null ? Array.Empty<MeshNode>() : new[] { placeholderRoot ?? root };
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
using System.Linq;
using System.Reactive.Linq;
using System.Text.Json;
using System.Threading.Tasks;
using MeshWeaver.Hosting.Persistence;
using MeshWeaver.Mesh;
using Xunit;

namespace MeshWeaver.Hosting.Test;

/// <summary>
/// Pins the CHILDREN INDEX behind <see cref="InMemoryStorageAdapter.ListChildPaths"/> (2026-09-13):
/// the listing used to scan every key of the store per call, quadratic in the mesh size for the
/// plugin gate's subtree queries. The index must answer exactly what the scan answered — direct
/// child NODES, and IMPLIED directories (a prefix with descendants but no node of its own) — and
/// follow deletes, version-gated writes and a dictionary seeded behind the adapter's back.
/// </summary>
public class InMemoryStorageAdapterChildIndexTest
{
private static readonly JsonSerializerOptions Options = new();

private static MeshNode Node(string path)
{
var slash = path.LastIndexOf('/');
return new MeshNode(path[(slash + 1)..], slash < 0 ? "" : path[..slash]) { Name = path, NodeType = "Markdown", State = MeshNodeState.Active };
}

private static async Task<(string[] Nodes, string[] Dirs)> ListAsync(InMemoryStorageAdapter adapter, string? parent)
{
var (nodes, dirs) = await adapter.ListChildPaths(parent).FirstAsync();
return (nodes.OrderBy(x => x).ToArray(), dirs.OrderBy(x => x).ToArray());
}

[Fact]
public async Task Lists_direct_nodes_and_implied_directories_and_follows_deletes()
{
var adapter = new InMemoryStorageAdapter();
foreach (var p in new[] { "Hosting", "Hosting/Build", "Hosting/Build/Source/BuildContent", "Store/Plugin", "Doc" })
await adapter.Write(Node(p), Options).FirstAsync();

var root = await ListAsync(adapter, null);
Assert.Equal(new[] { "Doc", "Hosting" }, root.Nodes);
Assert.Equal(new[] { "Store" }, root.Dirs); // Store has no node of its own, only a descendant

var hosting = await ListAsync(adapter, "Hosting");
Assert.Equal(new[] { "Hosting/Build" }, hosting.Nodes);
Assert.Empty(hosting.Dirs); // Build IS a node, so it is not an implied directory

var build = await ListAsync(adapter, "Hosting/Build");
Assert.Empty(build.Nodes);
Assert.Equal(new[] { "Hosting/Build/Source" }, build.Dirs);

await adapter.Delete("Hosting/Build/Source/BuildContent").FirstAsync();
Assert.Empty((await ListAsync(adapter, "Hosting/Build")).Dirs); // the implied directory went with its last descendant
Assert.True(await adapter.DeleteIfExists("Store/Plugin").FirstAsync());
Assert.Empty((await ListAsync(adapter, null)).Dirs); // and so did Store
}

[Fact]
public async Task Version_gated_writes_and_a_seeded_dictionary_are_indexed_too()
{
var nodes = new System.Collections.Concurrent.ConcurrentDictionary<string, MeshNode>(System.StringComparer.OrdinalIgnoreCase);
nodes["Seeded/Deep/Leaf"] = Node("Seeded/Deep/Leaf"); // behind the adapter's back
var adapter = new InMemoryStorageAdapter(nodes, new(System.StringComparer.OrdinalIgnoreCase));
Assert.Equal(new[] { "Seeded" }, (await ListAsync(adapter, null)).Dirs); // rebuilt on first use
Assert.True(await adapter.WriteIfVersion(Node("Seeded/Fresh"), 0, Options).FirstAsync());
Assert.Equal(new[] { "Seeded/Fresh" }, (await ListAsync(adapter, "Seeded")).Nodes);
Assert.Equal(new[] { "Seeded/Deep" }, (await ListAsync(adapter, "Seeded")).Dirs);
// a second adapter over the SAME dictionary sees the same index
var twin = new InMemoryStorageAdapter(nodes, new(System.StringComparer.OrdinalIgnoreCase));
Assert.Equal(new[] { "Seeded/Fresh" }, (await ListAsync(twin, "Seeded")).Nodes);
}
}
Loading