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
26 changes: 26 additions & 0 deletions src/MeshWeaver.Data/IReportsInitialLoadProgress.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
namespace MeshWeaver.Data;

/// <summary>
/// Optional capability of an <see cref="ITypeSource"/> whose initial load is itself more than one
/// wait, so the <see cref="DataContext"/> init time-box can say WHICH of them was outstanding when it
/// expired rather than only that the type source's leg was (Systemorph/MeshWeaver#1122).
///
/// <para>The canonical case is the per-node hub's <c>MeshNodeTypeSource</c>: its one leg is a
/// durable storage read CONCATENATED ahead of the routing-supplied own-node stream. A pending
/// <c>…/MeshNode</c> leg therefore did not separate "a storage read never came back" from "the
/// routing stream never delivered an acceptable node" — the one question the attributed timeout
/// exists to answer.</para>
///
/// <para><b>Diagnostic only, and it runs on a failure path.</b> An implementation reads
/// presence-only state it already holds: it must not create, block, subscribe or throw, and nothing
/// may ever branch on what it returns.</para>
/// </summary>
public interface IReportsInitialLoadProgress
{
/// <summary>
/// One short sentence naming the part of the initial load that is still outstanding, or
/// <c>null</c> when there is nothing more specific to say than the leg itself.
/// </summary>
/// <returns>The progress description, or <c>null</c>.</returns>
string? DescribeInitialLoadProgress();
}
43 changes: 36 additions & 7 deletions src/MeshWeaver.Data/TypeSourceInitializationLedger.cs
Original file line number Diff line number Diff line change
Expand Up @@ -31,13 +31,41 @@ namespace MeshWeaver.Data;
/// </summary>
internal sealed class TypeSourceInitializationLedger
{
private readonly ConcurrentDictionary<string, byte> pending = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, ITypeSource> pending = new(StringComparer.Ordinal);

/// <summary>
/// The unsettled legs, as <c>{streamId}/{collectionName}</c>. A snapshot — the fan-out keeps
/// running while a diagnostic reads this.
/// The unsettled legs, as <c>{streamId}/{collectionName}</c> — followed by
/// <c> [what inside the leg is outstanding]</c> when the type source implements
/// <see cref="IReportsInitialLoadProgress"/>. A snapshot — the fan-out keeps running while a
/// diagnostic reads this.
/// </summary>
internal IReadOnlyCollection<string> Pending => pending.Keys.ToArray();
internal IReadOnlyCollection<string> Pending => pending
.Select(kv => kv.Value is IReportsInitialLoadProgress progress
&& DescribeProgress(progress) is { Length: > 0 } detail
? $"{kv.Key} [{detail}]"
: kv.Key)
Comment on lines +42 to +46

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Real, and measured — fixed in ad5d0a4. TypeSourceInitializationLedger.DescribeProgress guards the reporter call and prints the fault in place of the sentence ([progress report FAULTED (InvalidOperationException: reporter broke)]), so it is surfaced in the message the operator reads, not hidden. New case in DataContextInitTimeoutNamesTheWaitInsideTheLegTest with a throwing reporter: with the guard removed the hub wedges (the probe times out behind a shut gate — the test's own wedge assertion fails); with it, the hub reaches FAILED and the fault is named. 5/5.

.ToArray();

/// <summary>
/// Asks an optional reporter for its sentence. This runs on the FAILURE path — inside
/// <c>SettleInitializationGate</c>, before the failure is recorded and the gate opened — so a
/// reporter that throws must not be able to take that path down with it and leave the hub
/// wedged instead of FAILED. The fault is not hidden: it is printed in place of the sentence,
/// naming the exception type, in the very message the operator reads.
/// </summary>
/// <param name="progress">The type source's reporter.</param>
/// <returns>Its sentence, or a sentence naming the fault it raised.</returns>
private static string? DescribeProgress(IReportsInitialLoadProgress progress)
{
try
{
return progress.DescribeInitialLoadProgress();
}
catch (Exception ex)
{
return $"progress report FAULTED ({ex.GetType().Name}: {ex.Message})";
}
}

/// <summary>
/// Marks every type source of <paramref name="stream"/> unsettled and hands back the claim the
Expand All @@ -53,9 +81,10 @@ internal StreamClaim Claim(
IEnumerable<ITypeSource> typeSources)
{
var streamId = stream.StreamId;
var keys = typeSources.Select(ts => Key(streamId, ts)).ToArray();
foreach (var key in keys)
pending[key] = 0;
var legs = typeSources.Select(ts => (Key: Key(streamId, ts), Source: ts)).ToArray();
foreach (var leg in legs)
pending[leg.Key] = leg.Source;
var keys = legs.Select(l => l.Key).ToArray();

// Bounds growth on a data source whose partition streams churn: without this, every
// stream that is torn down before settling leaves its legs behind forever.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,33 @@ The ledger is diagnostic only: nothing waits on it and nothing branches on it. T
is unchanged, so the existing incident keeps collecting its history while every sample line now
carries the cause.

### 🚨 For a per-node hub the leg alone did NOT separate those two — it now says which half

The claim just above — that the leg "separates a storage read never came back from a remote hub
never answered" — was not true for the case that matters most. A per-node hub has ONE leg, `…/MeshNode`,
and that leg is `MeshNodeTypeSource.Initialize`: the durable read **concatenated ahead of** the
routing-supplied own-node stream (above). The first attributed production sample after the ledger
shipped, `Collaboration` on memex-cloud (2026-09-21 17:06:24Z, Systemorph/MeshWeaver#1122), read
*"type-source legs still outstanding: 7j8ehN2m0UCo51iBcrGL2A/MeshNode"* — and could be either.

A type source that implements `IReportsInitialLoadProgress` now has its own sentence appended to its
leg, and `MeshNodeTypeSource` reports which half is outstanding:

| sentence after the leg | where to look |
|---|---|
| `[durable seed read of 'P' outstanding for 115.0s — a storage read that has not come back]` | storage: the `pg-read` gate's wait, a wedged adapter |
| `[durable seed read of 'P' found no row after 0.1s; the routing-supplied own-node stream has not emitted]` | the routing / stream-cache side — the read is done and the node never arrived |
| `[… ; the routing-supplied own-node stream emitted N time(s) and none was accepted]` | the own-node gate dropped every emission (a null, or a stale version) |
| `[durable seed read of 'P' FAULTED (…) after …]` | the read faulted and degraded to the routing leg, which then did not deliver |

Still diagnostic only: the progress is written from the load's own callbacks and read by the
failure path; nothing waits on it. Pinned by `DataContextInitTimeoutNamesTheWaitInsideTheLegTest`
(the rendering — red with the ledger printing keys only) and `MeshNodeTypeSourceInitialLoadProgressTest`
(the sentences, and the `TrackSeed`/`TrackRouting` wiring `Initialize` composes, driven with
controllable streams). A reporter that throws is printed as `[progress report FAULTED (…)]` rather
than escaping: the rendering runs inside `SettleInitializationGate`, and with that guard removed the
hub wedges instead of reaching FAILED — measured, it is the second assertion of the Data test.

## A failed init errors every stream it holds, and creates none

The failure used to be propagated with `ds.GetStreamForPartition(null).OnError(failure)`. That
Expand Down
117 changes: 113 additions & 4 deletions src/MeshWeaver.Graph/MeshNodeTypeSource.cs
Original file line number Diff line number Diff line change
Expand Up @@ -44,8 +44,12 @@ namespace MeshWeaver.Graph;
/// announce-loss class. Deletes have no such ambiguity (a path leaving the collection is always a
/// genuine delete) and always announce via <see cref="StorageAdapterChangeFeedExtensions.DeleteAndPublish"/>.</para>
/// </summary>
public record MeshNodeTypeSource : TypeSourceWithType<MeshNode, MeshNodeTypeSource>
public record MeshNodeTypeSource : TypeSourceWithType<MeshNode, MeshNodeTypeSource>, IReportsInitialLoadProgress
{
// Which half of Initialize's Concat is outstanding — read only by the DataContext init
// time-box's diagnostic (IReportsInitialLoadProgress). A reference, so the record's `with`
// copies share it, exactly as the data source's initialization ledger is shared.
private readonly InitialLoadProgress _initialLoad = new();
private readonly IStorageAdapter? _persistenceCore;
private readonly string _hubPath; // e.g., "graph/org1"
private readonly IWorkspace _workspace;
Expand Down Expand Up @@ -843,7 +847,8 @@ protected override IObservable<InstanceCollection> Initialize(
// HubConfiguration delegate that the compile watcher / first-build kickoff /
// data-model areas read off the workspace's own node.
return DurableSeed().Select(n => (Node: GraftRoutingEnrichment(n), Durable: true))
.Concat(_ownNodeStream.Select(n => (Node: n, Durable: false)))
.Concat(_initialLoad.TrackRouting(_ownNodeStream)
.Select(n => (Node: n, Durable: false)))
.Where(e => AcceptOwnNodeEmission(e.Node, e.Durable))
.Select(e => BuildInstanceCollection(e.Node));

Expand Down Expand Up @@ -928,8 +933,7 @@ protected override IObservable<InstanceCollection> Initialize(
if (_persistenceCore is null)
return Observable.Empty<MeshNode?>();

return _persistenceCore.Read(_hubPath, _workspace.Hub.JsonSerializerOptions)
.Take(1)
return _initialLoad.TrackSeed(_persistenceCore.Read(_hubPath, _workspace.Hub.JsonSerializerOptions))
.Do(seed =>
{
_logger?.LogDebug(
Expand Down Expand Up @@ -1193,4 +1197,109 @@ private MeshNode ResolveJsonElementContent(MeshNode node)

return node;
}

/// <inheritdoc />
/// <remarks>
/// Separates the two waits <see cref="Initialize"/> concatenates, which a pending
/// <c>…/MeshNode</c> leg alone cannot: the unbounded durable storage read of this hub's own row,
/// and — only once that settled — the routing-supplied own-node stream. "The read has been
/// outstanding for 115 s" points at storage (the <c>pg-read</c> gate, a wedged adapter); "the read
/// settled after 0.1 s and the routing stream has not emitted" points at the routing and
/// stream-cache side. Those are opposite investigations, and before this the timeout could not say
/// which one to start (Systemorph/MeshWeaver#1122).
/// </remarks>
public string? DescribeInitialLoadProgress() =>
_initialLoad.Describe(_hubPath, _ownNodeStream is not null);

/// <summary>
/// Presence-only record of how far <see cref="Initialize"/>'s initial load got. Written from the
/// load's own callbacks and read only by <see cref="DescribeInitialLoadProgress"/>; nothing waits
/// on it and nothing branches on it.
/// </summary>
internal sealed class InitialLoadProgress
{
/// <summary>One immutable reading of the durable seed. Published by a single reference
/// swap, so a concurrent <see cref="Describe"/> sees either the whole settlement or none of
/// it — never a settle time without its outcome.</summary>
private sealed record SeedState(long StartedTicks, long SettledTicks, string? Outcome);

private SeedState? seed;
private int routingEmissions;

/// <summary>
/// Wraps the durable read so its subscription and settlement are recorded — the one place
/// <see cref="Initialize"/>'s seed wiring lives, so a test can drive it with a controllable
/// read. The FIRST settle wins: <c>Take(1)</c> completes right after its value, and that
/// completion must not overwrite "read version N".
/// </summary>
/// <param name="read">The durable read of this hub's own row.</param>
/// <returns>The read, taking at most one value, with its progress recorded.</returns>
public IObservable<MeshNode?> TrackSeed(IObservable<MeshNode?> read) =>
Observable.Defer(() =>
{
SeedReadStarted();
return read;
})
.Take(1)
.Do(
node => SeedSettled(node is null ? "found no row" : "read version " + node.Version),
ex => SeedSettled("FAULTED (" + ex.GetType().Name + ")"),
() => SeedSettled("completed with no row"));

/// <summary>Wraps the routing-supplied own-node stream so each emission is counted.</summary>
/// <typeparam name="T">The routing stream's element type.</typeparam>
/// <param name="routing">The routing-supplied own-node stream.</param>
/// <returns>The same stream, counted.</returns>
public IObservable<T> TrackRouting<T>(IObservable<T> routing) =>
routing.Do(_ => RoutingEmitted());

/// <summary>The durable read was subscribed.</summary>
public void SeedReadStarted() =>
Volatile.Write(ref seed, new SeedState(DateTimeOffset.UtcNow.UtcTicks, 0, null));

/// <summary>The durable read settled; only the first settle after a start is kept.</summary>
/// <param name="outcome">How it settled.</param>
public void SeedSettled(string outcome)
{
var current = Volatile.Read(ref seed);
if (current is null || current.SettledTicks != 0)
return;
Interlocked.CompareExchange(
ref seed, current with { SettledTicks = DateTimeOffset.UtcNow.UtcTicks, Outcome = outcome }, current);
}

/// <summary>The routing-supplied own-node stream produced a value (accepted or not).</summary>
public void RoutingEmitted() => Interlocked.Increment(ref routingEmissions);

/// <summary>Renders the progress, or <c>null</c> when the load never reached either wait.</summary>
/// <param name="hubPath">This hub's path.</param>
/// <param name="concatenatesRoutingStream">Whether the load has a routing leg at all.</param>
/// <returns>The description.</returns>
public string? Describe(string hubPath, bool concatenatesRoutingStream)
{
var emissions = Volatile.Read(ref routingEmissions);
var routing = emissions == 0
? "the routing-supplied own-node stream has not emitted"
: $"the routing-supplied own-node stream emitted {emissions} time(s) and none was accepted";

var state = Volatile.Read(ref seed);
if (state is null)
return concatenatesRoutingStream ? $"no durable seed read; {routing}" : null;

var started = state.StartedTicks;
var settled = state.SettledTicks;
if (settled == 0)
{
var outstanding = TimeSpan.FromTicks(DateTimeOffset.UtcNow.UtcTicks - started);
return FormattableString.Invariant(
$"durable seed read of '{hubPath}' outstanding for {outstanding.TotalSeconds:0.0}s ")
+ "— a storage read that has not come back";
}

var readFor = TimeSpan.FromTicks(settled - started);
var head = FormattableString.Invariant(
$"durable seed read of '{hubPath}' {state.Outcome} after {readFor.TotalSeconds:0.0}s");
return concatenatesRoutingStream ? $"{head}; {routing}" : head;
}
}
}
Loading
Loading