From fb8ddbf4c6162c4a3adc062e18afac3922dd9043 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= Date: Thu, 13 Aug 2026 09:52:05 +0200 Subject: [PATCH 1/2] feat: bound an activity's message list, flush the overflow to segment satellites MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An activity's transcript lived entirely in `ActivityLog.Messages`, and every `stream.Update` re-serialises the whole `MeshNode.Content` to compute its patch — so N appends onto one growing list cost O(N²). On memex-cloud a single import activity spent ~719 MB of serialisation across 5,239 writes to a 141 kB node (~7.1 GB across 25,579 activity nodes), the dominant term in that pod's CFS throttling. No delta field escapes that. The cross-hub path ships an RFC 7396 merge patch, which clones a changed array whole, and `ExtractBaseValues` clones the previous array too — one append to an N-element collection ships ~2N elements, and `BaseValues` is genuinely consumed by `MergeArray`, so it cannot just be dropped. Bounding the head is the only lever that changes the asymptotics. `Messages` is now a window of at most `MessageWindowLimit` (500). Older lines are sealed into `ActivityLogSegment` satellites at `{activityPath}/_Log/ {index:D6}` — same `activities` table as the activity itself, no new table — leaving `MessageWindowKeep` (100) on the head. `ActivityLogAppender` is the single write surface: it appends and applies a terminal-status change in ONE update, so a reader can never see the status before the lines explaining it. An activity that never reaches 500 messages behaves byte-identically to before — same single write per append. Only long activities take the new path, which is why ~40 existing activity tests are untouched. MEASURED, on 500 vs 2,000 messages into one activity (StaticRepoImportActivityWriteCountTest.AppendCost_DoesNotGrowWithTheLengthOfTheActivity): before after head bytes 88,729 → 353,994 88,711 → 53,366 head-size ratio 3.99× 0.60× cumulative serialised 0.93 MB → 14.3 MB 0.93 MB → 3.98 MB writes (80 append calls) 80 84 (+1 trim per segment) 4× the messages cost 15.4× the serialisation before and 4.3× after — O(N²) to O(N). The test itself runs 26 s → 10 s for the same work. The flush is lock-free and loses nothing. A seal is CLAIMED inside the head's update lambda, which the owning hub serialises, so exactly one appender claims each slice and no two overlap. The claimed messages stay on the head until the segment write succeeds; only then are they trimmed. A crash or a failed segment write therefore needs no watchdog — the claim is still standing and the messages are still there, so the next append retries the same slice against the same deterministic index. `ActivityLogLogger` (the kernel script logger) re-asserts whole content every 100 ms rather than patching, so it is the single writer of its node and seals its own overflow directly under the lock it already holds for the terminal settle — same window, same segments, no claim protocol needed. Its terminal settle is otherwise untouched. Also fixes a serialisation bug the measurement exposed: System.Text.Json serialises get-only properties, so the derived `TotalMessageCount` reached the merge patch as a field no writer had set — a phantom conflict the merge guard refused ("changed since the writer's base"), taking real fields in the same patch down with it. Every computed property here is now `[JsonIgnore]`; without it `SealedMessages` would also have serialised a 400-message list into exactly the write the window exists to shrink. Readers: `TotalMessageCount` replaces `Messages.Count` wherever a total was meant; the log pane says how many earlier messages were archived. Tail readers (`[^1]`, `TakeLast(n)`) are unaffected — the window keeps the most recent lines. Segments are enumerated with a children query, never a point-read of a possibly-absent satellite. Co-Authored-By: Claude Opus 5 (1M context) --- clients/react/src/i18n/strings.de.json | 1 + clients/react/src/i18n/strings.en.json | 1 + .../ContentIndexingActivity.cs | 43 ++--- src/MeshWeaver.Data.Contract/ActivityLog.cs | 106 ++++++++++- .../ActivityLogSegment.cs | 44 +++++ src/MeshWeaver.Data/DataExtensions.cs | 3 + .../Data/Architecture/ActivityOperations.md | 43 +++++ src/MeshWeaver.GitSync/ActivityRunner.cs | 56 +++--- src/MeshWeaver.Graph/ActivityLayoutAreas.cs | 13 +- .../ActivityLogSegmentNodeType.cs | 58 ++++++ .../GraphConfigurationExtensions.cs | 1 + .../NodeTypeCompilationActivity.cs | 122 ++++--------- .../ActivityLogLogger.cs | 98 +++++++++- src/MeshWeaver.Mcp/McpMeshPlugin.cs | 2 +- .../ActivityLogAppender.cs | 169 ++++++++++++++++++ .../SatelliteTableMapping.cs | 5 + .../Localization/strings.de.json | 1 + .../Localization/strings.en.json | 1 + .../StaticRepoImportActivityWriteCountTest.cs | 166 +++++++++++++++++ 19 files changed, 782 insertions(+), 151 deletions(-) create mode 100644 src/MeshWeaver.Data.Contract/ActivityLogSegment.cs create mode 100644 src/MeshWeaver.Graph/Configuration/ActivityLogSegmentNodeType.cs create mode 100644 src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs diff --git a/clients/react/src/i18n/strings.de.json b/clients/react/src/i18n/strings.de.json index 7360128c63..63f470df02 100644 --- a/clients/react/src/i18n/strings.de.json +++ b/clients/react/src/i18n/strings.de.json @@ -412,6 +412,7 @@ "ui.sourceChanged": "Quelle geändert — muss kompiliert werden", "ui.upToDate": "Aktuell", "ui.notYetRun": "Noch nicht ausgeführt.", + "ui.activityEarlierMessages": "… {0} frühere Meldungen archiviert", "ui.activityNoOutput": "Dieser Lauf hat keine Ausgabe erzeugt.", "ui.running": "Wird ausgeführt…", "ui.statusDone": "Fertig", diff --git a/clients/react/src/i18n/strings.en.json b/clients/react/src/i18n/strings.en.json index 25159eeeb8..48b01e9d91 100644 --- a/clients/react/src/i18n/strings.en.json +++ b/clients/react/src/i18n/strings.en.json @@ -412,6 +412,7 @@ "ui.sourceChanged": "Source changed — needs compile", "ui.upToDate": "Up to date", "ui.notYetRun": "Not yet run.", + "ui.activityEarlierMessages": "… {0} earlier messages archived", "ui.activityNoOutput": "This run produced no output.", "ui.running": "Running…", "ui.statusDone": "Done", diff --git a/src/MeshWeaver.ContentCollections.Indexing.Graph/ContentIndexingActivity.cs b/src/MeshWeaver.ContentCollections.Indexing.Graph/ContentIndexingActivity.cs index f3fe4c603f..e792bd8549 100644 --- a/src/MeshWeaver.ContentCollections.Indexing.Graph/ContentIndexingActivity.cs +++ b/src/MeshWeaver.ContentCollections.Indexing.Graph/ContentIndexingActivity.cs @@ -210,17 +210,12 @@ private static void Append( // context-null → the partition RLS denies → the progress line never lands. using (owner is not null && accessService is not null ? accessService.SwitchAccessContext(owner) : null) { - workspace.GetMeshNodeStream(activityPath).Update(node => - { - // ContentAs, NOT `Content as ActivityLog`: the activity node is owned by the partition's - // _Activity hub, so this cross-hub Update lambda diffs against a LOCAL mirror whose Content - // can be a JsonElement (not strongly-typed). A plain cast would be null → the update no-ops → - // the RFC 7396 patch is never sent. ContentAs deserializes the JsonElement so the write lands. - // (project_baddata_contentas_pattern.) - var log = node.ContentAs(workspace.Hub.JsonSerializerOptions, logger); - if (log is null) return node; - return node with { Content = log.Append(messages) }; - }).Subscribe(_ => { }, ex => logger?.LogWarning(ex, "Indexing activity log append failed for {Path}", activityPath)); + // ActivityLogAppender, not a hand-rolled Update: it keeps Messages bounded and flushes the + // overflow into _Log segment satellites, so a full reindex's per-batch progress costs + // O(batches) instead of O(batches²) in re-serialisation. + ActivityLogAppender + .Append(workspace.Hub, activityPath, messages, mutate: null, logger) + .Subscribe(_ => { }, ex => logger?.LogWarning(ex, "Indexing activity log append failed for {Path}", activityPath)); } } @@ -233,23 +228,15 @@ private static IObservable Finish( // RLS lets the write land — otherwise the activity hangs Running forever. using (owner is not null && accessService is not null ? accessService.SwitchAccessContext(owner) : null) { - return workspace.GetMeshNodeStream(activityPath).Update(node => - { - // ContentAs, NOT `Content as ActivityLog`: the cross-hub Update lambda diffs against a - // local mirror whose Content may be a JsonElement; a plain cast would no-op and the terminal - // Status would never be written → the activity hangs Running forever (the ReindexAll / Upload - // indexing-activity timeout). See project_baddata_contentas_pattern. - var log = node.ContentAs(workspace.Hub.JsonSerializerOptions, logger); - if (log is null) return node; - var withFinal = string.IsNullOrEmpty(finalMessage) - ? log - : log.Append(new LogMessage(finalMessage, - status == ActivityStatus.Failed ? LogLevel.Error : LogLevel.Information)); - return node with - { - Content = withFinal with { Status = status, End = DateTime.UtcNow, RequestedStatus = null } - }; - }).Select(_ => Unit.Default); + return ActivityLogAppender.Append( + workspace.Hub, activityPath, + string.IsNullOrEmpty(finalMessage) + ? [] + : [new LogMessage(finalMessage, + status == ActivityStatus.Failed ? LogLevel.Error : LogLevel.Information)], + log => log with { Status = status, End = DateTime.UtcNow, RequestedStatus = null }, + logger) + .Select(_ => Unit.Default); } } diff --git a/src/MeshWeaver.Data.Contract/ActivityLog.cs b/src/MeshWeaver.Data.Contract/ActivityLog.cs index 8ac09dd8ce..934575e474 100644 --- a/src/MeshWeaver.Data.Contract/ActivityLog.cs +++ b/src/MeshWeaver.Data.Contract/ActivityLog.cs @@ -1,6 +1,7 @@ using System.Collections.Immutable; using System.ComponentModel.DataAnnotations; using System.Text.Json; +using System.Text.Json.Serialization; using Microsoft.Extensions.Logging; using MeshWeaver.ShortGuid; @@ -20,12 +21,58 @@ public record ActivityLog(string Category) /// The workspace version recorded when the activity finished. public int Version { get; init; } + /// + /// How many messages the head keeps before sealing the oldest ones into an + /// satellite. An activity that never exceeds this behaves + /// exactly as it did before segments existed — same single write per append, same bytes. + /// + public const int MessageWindowLimit = 500; + + /// + /// How many messages stay on the head after a seal. The gap to + /// is what amortises the seal: one extra write and one segment + /// create per Limit - Keep appends. Every live reader wants far less than this — the + /// widest is a 12-line preview — so nothing on screen notices the trim. + /// + public const int MessageWindowKeep = 100; + /// Unique identifier of this activity. [property: Key] public string Id { get; init; } = Guid.NewGuid().AsString(); - /// The log messages accumulated by this activity, in order. + + /// + /// The activity's log messages, most recent last — a bounded WINDOW, not necessarily the whole + /// transcript. Once an activity exceeds the oldest lines are + /// sealed into satellites under {activityPath}/_Log and + /// drop out of here, so a write stays O(1) instead of O(transcript). + /// + /// 🚨 This is the reason nothing may derive a total, a severity, or a progress signal from + /// Messages.Count — use and . + /// The window still holds the LAST messages, so tail readers ([^1], TakeLast(n)) + /// are unaffected. + /// public ImmutableList Messages { get; init; } = ImmutableList.Empty; + /// + /// How many satellites this activity has sealed — also the index + /// of the next one. A reader enumerates the segments with a children query rather than trusting + /// this count, so a seal whose satellite write failed leaves a harmless gap, never a dangling + /// point-read. + /// + public int SegmentCount { get; init; } + + /// + /// How many LEADING entries of a seal has claimed but not yet trimmed — + /// the flush's claim marker, written atomically with the append that made it. + /// + /// 🚨 This is what makes a flush safe without a lock. Writes to one node are serialised by + /// the owning hub, so exactly one appender can move this 0 → n; every other appender sees it + /// non-zero and does not claim a second, overlapping slice. The claimed messages stay in + /// until the segment is durable, so a crash or a failed segment write + /// loses nothing — the next append simply retries the same slice. + /// + public int SealingCount { get; init; } + /// /// How many messages this activity has appended in total — plus anything /// no longer held there. Maintained incrementally by . @@ -93,7 +140,12 @@ public record ActivityLog(string Category) /// How many messages this activity has recorded, counting any no longer held in /// . Back-compatible with logs persisted before /// existed (those carry 0, so the in-list count wins). + /// 🚨 [JsonIgnore], like every computed property here. System.Text.Json serialises + /// get-only properties by default, and anything that reaches the JSON reaches the RFC 7396 merge + /// patch — where a derived field the writer never set becomes a phantom conflict the merge guard + /// refuses ("changed since the writer's base"), taking the real fields in that patch down with it. /// + [JsonIgnore] public int TotalMessageCount => Math.Max(MessageCount, Messages.Count); /// @@ -126,6 +178,58 @@ public ActivityLog Append(IReadOnlyList messages) /// The activity log with the message appended and the roll-ups advanced. public ActivityLog Append(LogMessage message) => Append([message]); + /// + /// Claims the oldest messages for sealing into an when the window + /// has overflowed — a no-op below and while another seal is still + /// in flight. Nothing is removed here: the claim only marks how many LEADING entries of + /// the next segment will carry, so the messages remain readable on the head + /// until the satellite is durable. + /// Called inside the head's update lambda, which the owning hub serialises — that is what + /// makes "exactly one appender claims each slice" true without a lock. + /// + /// The activity log with the claim recorded, or the same instance when nothing to claim. + public ActivityLog ClaimSeal() => + SealingCount == 0 && Messages.Count > MessageWindowLimit + ? this with { SealingCount = Messages.Count - MessageWindowKeep } + : this; + + /// + /// The messages a claimed seal must write into its segment — the leading + /// entries of . Empty when no seal is claimed. + /// + /// 🚨 [JsonIgnore] — without it this list would be serialised into every patch + /// alongside , doubling the payload of exactly the write the window exists + /// to shrink. + [JsonIgnore] + public ImmutableList SealedMessages => + SealingCount <= 0 ? ImmutableList.Empty : Messages.GetRange(0, Math.Min(SealingCount, Messages.Count)); + + /// + /// The activity-wide ordinal of the first message in the currently claimed seal — what the segment + /// records as its . + /// + [JsonIgnore] + public int SealedFirstOrdinal => TotalMessageCount - Messages.Count; + + /// + /// Drops the claimed messages from the window now that their segment is durable, and advances + /// . Idempotent: a no-op when no seal is claimed, so a retried completion + /// cannot trim a second, unrelated slice. + /// Trimming from the FRONT is what makes this order-independent: messages only ever arrive at + /// the end, so the leading entries are still exactly the ones the claim + /// captured, however many appends landed in between. + /// + /// The activity log with the sealed slice removed. + public ActivityLog CompleteSeal() => + SealingCount <= 0 + ? this + : this with + { + Messages = Messages.RemoveRange(0, Math.Min(SealingCount, Messages.Count)), + SegmentCount = SegmentCount + 1, + SealingCount = 0, + }; + /// /// Returns a copy of this activity marked with the /// given error appended as an error message and set to now. diff --git a/src/MeshWeaver.Data.Contract/ActivityLogSegment.cs b/src/MeshWeaver.Data.Contract/ActivityLogSegment.cs new file mode 100644 index 0000000000..ad12b9a427 --- /dev/null +++ b/src/MeshWeaver.Data.Contract/ActivityLogSegment.cs @@ -0,0 +1,44 @@ +using System.Collections.Immutable; +using System.ComponentModel.DataAnnotations; +using MeshWeaver.ShortGuid; + +namespace MeshWeaver.Data; + +/// +/// A sealed slice of an activity's transcript, stored as a satellite MeshNode at +/// {activityPath}/_Log/{index:D6} once the messages have scrolled out of +/// (the head's bounded window). +/// +/// Why the transcript moves off the head. Every stream.Update re-serialises the +/// WHOLE MeshNode.Content to compute its patch, so appending N lines to one growing list costs +/// O(N²) — on memex-cloud a single import activity spent ~719 MB of serialisation across 5,239 writes +/// to a 141 kB node. No delta field fixes that: the cross-hub path ships an RFC 7396 merge patch, which +/// clones a changed array whole, and the three-way merge's base extraction clones the previous array +/// too. Bounding the head is what changes the asymptotics — with the window fixed, each write is O(1) +/// and the whole activity is O(N). +/// +/// Segments are append-once and immutable. A segment is written exactly once, by the +/// append that sealed it, and never updated afterwards — so nothing ever reads-then-writes a segment +/// path, and no reader may point-read one that might be absent (that opens the shared stream cache's +/// storm breaker on a path a concurrent write is about to use). Enumerate them with a children query +/// on {activityPath}/_Log; the ordering key is . +/// +public record ActivityLogSegment +{ + /// Unique identifier of this segment — mirrors the node id. + [property: Key] + public string Id { get; init; } = Guid.NewGuid().AsString(); + + /// + /// The activity-wide ordinal of this segment's FIRST message: how many messages the activity had + /// already recorded before this slice. Segments are ordered by it, and it is what lets a reader + /// stitch segments and the head window back into the original sequence. + /// + public int FirstOrdinal { get; init; } + + /// The messages in this slice, in order. + public ImmutableList Messages { get; init; } = ImmutableList.Empty; + + /// The path of the activity this slice belongs to. + public string? ActivityPath { get; init; } +} diff --git a/src/MeshWeaver.Data/DataExtensions.cs b/src/MeshWeaver.Data/DataExtensions.cs index 67a9fc7a3c..fb60fb8965 100644 --- a/src/MeshWeaver.Data/DataExtensions.cs +++ b/src/MeshWeaver.Data/DataExtensions.cs @@ -199,6 +199,9 @@ private static MessageHubConfiguration GetDefaultConfiguration(MessageHubConfigu ) .WithType(typeof(Address), nameof(Address)) .WithType(typeof(ActivityLog), nameof(ActivityLog)) + // A sealed slice of an activity's transcript, stored as its own satellite node. Registered + // beside ActivityLog because the segment crosses the wire on exactly the same paths. + .WithType(typeof(ActivityLogSegment), nameof(ActivityLogSegment)) // ActivityLog content children — serialised inside log entries / user attribution. .WithType(typeof(LogMessage), nameof(LogMessage)) .WithType(typeof(UserInfo), nameof(UserInfo)) diff --git a/src/MeshWeaver.Documentation/Data/Architecture/ActivityOperations.md b/src/MeshWeaver.Documentation/Data/Architecture/ActivityOperations.md index 80e29cfade..916b2ef8da 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/ActivityOperations.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/ActivityOperations.md @@ -159,6 +159,49 @@ The running script receives the cancellation, throws `OperationCanceledException --- +## Writing log messages: `ActivityLogAppender` + +Every log line written onto a **persisted activity node** goes through `ActivityLogAppender.Append` (`src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs`). It is the append-side twin of the transition surface above: callers hand it messages plus an optional change to the log (terminal status, `End`, `ReturnValue`) and it performs **one** `stream.Update` carrying both — so a reader can never observe the terminal status before the lines that explain it. + +```csharp +ActivityLogAppender.Append(hub, activityPath, [new LogMessage(text, LogLevel.Information)]) + .Subscribe(_ => { }, ex => logger.LogDebug(ex, "activity append failed")); + +// terminal status + its explanation, atomically: +ActivityLogAppender.Append(hub, activityPath, [new LogMessage(error, LogLevel.Error)], + log => log with { Status = ActivityStatus.Failed, End = DateTime.UtcNow }) + .Subscribe(_ => { }, ex => logger.LogDebug(ex, "activity complete failed")); +``` + +### `Messages` is a bounded window + +`ActivityLog.Messages` holds at most `ActivityLog.MessageWindowLimit` (500) entries. Older lines are sealed into `ActivityLogSegment` satellites at `{activityPath}/_Log/{index:D6}` and drop off the head, leaving `MessageWindowKeep` (100) behind. + +**Why.** Every `stream.Update` re-serialises the whole `MeshNode.Content` to compute its patch, so appending N lines to one growing list costs **O(N²)** — measured at ~719 MB of serialisation for a single memex-cloud import activity (5,239 writes over a 141 kB node), and the dominant term in that pod's CFS throttling. A delta field does not escape it: the cross-hub path ships an RFC 7396 merge patch, which clones a changed array whole, and the three-way merge's base extraction clones the previous array too — so one append to an N-element collection ships ~2N elements. Bounding the head is the only lever that changes the asymptotics; with the window fixed, each write is O(1) and the activity is O(N). + +**Below the window nothing changes.** An activity that never reaches 500 messages takes exactly the single write per append it always did, with byte-identical content. Only long activities — the ones that actually hurt — take the new path. + +### Consequences for readers + +| You want… | Read… | +|---|---| +| How many lines the activity produced | `log.TotalMessageCount` — **never** `log.Messages.Count` | +| Whether it errored / its terminal status | `log.HasErrors()`, `log.Finish(...)` — both answer from the `MaxSeverity` counter | +| The latest line, or the last few | `log.Messages` — the window keeps the **most recent** entries, so `[^1]` and `TakeLast(n)` are unaffected | +| The full transcript | the window plus the `_Log` segments, ordered by `ActivityLogSegment.FirstOrdinal` | + +🚨 **Enumerate segments with a children query on `{activityPath}/_Log`, never a point-read of a segment path.** A point-read of an absent satellite opens the shared stream cache's storm breaker on a path a concurrent write is about to use, and the breaker fast-fails writes too. + +🚨 **Never derive progress from `Messages.Count`.** It stops growing once an activity passes the window, so anything keyed on it — a `DistinctUntilChanged`, a change detector — silently freezes for exactly the long-running activities it exists to follow. `TotalMessageCount` is the monotonic signal. + +### How the flush stays safe without a lock + +A seal is **claimed inside the head's update lambda** (`ActivityLog.ClaimSeal`), which the owning hub serialises — so exactly one appender claims each slice and no two claims overlap. The claimed messages **stay on the head** until the segment write succeeds; only then are they trimmed (`ActivityLog.CompleteSeal`). A crash or a failed segment write therefore loses nothing and needs no watchdog: the claim is still standing and the messages are still there, so the next append retries the identical slice against the same (deterministic) segment index. + +`ActivityLogLogger` (the kernel's script logger) is the one writer that does **not** use the appender: it re-asserts whole content on every 100 ms flush rather than patching, so it is the single writer of its node and seals its own overflow directly under the lock it already holds for the terminal settle. Same window, same segment shape, no claim protocol needed because there is no second appender to race. + +--- + ## See also - [Activity Control Plane](/Doc/Architecture/ActivityControlPlane) — the `Status` / `RequestedStatus` pattern and how to wire your own NodeType to it diff --git a/src/MeshWeaver.GitSync/ActivityRunner.cs b/src/MeshWeaver.GitSync/ActivityRunner.cs index bdbe63c742..fefac48ca9 100644 --- a/src/MeshWeaver.GitSync/ActivityRunner.cs +++ b/src/MeshWeaver.GitSync/ActivityRunner.cs @@ -248,16 +248,14 @@ private static void Append( // the cross-hub patch posts context-null → the partition's RLS denies → the // progress line never lands. The owner can Update the partition (the satellite // rule delegates there), so the write succeeds under their identity. + // ActivityLogAppender, not a hand-rolled Update: it keeps Messages bounded and flushes the + // overflow into _Log segment satellites, so a long-running sync's per-item progress costs + // O(items) instead of O(items²) in re-serialisation. using (owner is not null && accessService is not null ? accessService.SwitchAccessContext(owner) : null) { - workspace.GetMeshNodeStream(activityPath).Update(node => - { - // Tolerant read (ContentAs), not a bare type test: on a cross-hub update the - // content can arrive as a degraded JsonElement — `is ActivityLog` would silently - // no-op and the progress line would never land (same read as the cancel watch). - if (node.ContentAs(workspace.Hub.JsonSerializerOptions, logger) is not { } log) return node; - return node with { Content = log.Append(new LogMessage(message, level)) }; - }).Subscribe(_ => { }, ex => logger?.LogWarning(ex, "Activity log append failed for {Path}", activityPath)); + ActivityLogAppender + .Append(workspace.Hub, activityPath, [new LogMessage(message, level)], mutate: null, logger) + .Subscribe(_ => { }, ex => logger?.LogWarning(ex, "Activity log append failed for {Path}", activityPath)); } } @@ -271,30 +269,24 @@ private static IObservable Finish( // Running forever. using (owner is not null && accessService is not null ? accessService.SwitchAccessContext(owner) : null) { - return workspace.GetMeshNodeStream(activityPath).Update(node => - { - // Tolerant read (ContentAs), not a bare type test: on a cross-hub update the - // content can arrive as a degraded JsonElement — `is ActivityLog` would silently - // no-op and the activity would hang Running forever (same read as the cancel watch). - if (node.ContentAs(workspace.Hub.JsonSerializerOptions, logger) is not { } log) return node; - var withFinal = string.IsNullOrEmpty(finalMessage) - ? log - : log.Append(new LogMessage(finalMessage, - status == ActivityStatus.Failed ? LogLevel.Error : LogLevel.Information)); - // Honour what the command reported via ctx.Log: ActivityLog.Finish computes - // MAX(status, roll-up from Messages) — an Error line a command appended flips - // a would-be Succeeded terminal to Failed, a Warning line to Warning; explicit - // Failed / Cancelled always stand (they are more severe in the enum order). - // Without this, a command that reports per-item errors but completes without - // throwing would end "Succeeded" with errors in its own log. - // The hub's current version stamps the finished log — the same - // `(int)hub.Version` every other Finish caller records. - var finished = withFinal.Finish((int)workspace.Hub.Version, status); - return node with - { - Content = finished with { RequestedStatus = null } - }; - }).Select(_ => Unit.Default); + return ActivityLogAppender.Append( + workspace.Hub, activityPath, + string.IsNullOrEmpty(finalMessage) + ? [] + : [new LogMessage(finalMessage, + status == ActivityStatus.Failed ? LogLevel.Error : LogLevel.Information)], + // Honour what the command reported via ctx.Log: ActivityLog.Finish computes + // MAX(status, severity roll-up) — an Error line a command appended flips a + // would-be Succeeded terminal to Failed, a Warning line to Warning; explicit + // Failed / Cancelled always stand (they are more severe in the enum order). + // Without this, a command that reports per-item errors but completes without + // throwing would end "Succeeded" with errors in its own log. The roll-up is a + // counter, so this stays correct once older lines have left the window. + // The hub's current version stamps the finished log — the same + // `(int)hub.Version` every other Finish caller records. + log => log.Finish((int)workspace.Hub.Version, status) with { RequestedStatus = null }, + logger) + .Select(_ => Unit.Default); } } } diff --git a/src/MeshWeaver.Graph/ActivityLayoutAreas.cs b/src/MeshWeaver.Graph/ActivityLayoutAreas.cs index 8ee1ca7106..8fa472437e 100644 --- a/src/MeshWeaver.Graph/ActivityLayoutAreas.cs +++ b/src/MeshWeaver.Graph/ActivityLayoutAreas.cs @@ -360,9 +360,20 @@ public static StackControl BuildLog(ActivityLog log, string? locale = null, bool "font-family: var(--font-monospace, ui-monospace, monospace); " + "font-size: .85rem; gap: 2px; max-height: 320px; overflow: auto;"); - if (log.Messages.Count == 0) + if (log.TotalMessageCount == 0) return hasResult ? stack : stack.WithView(BuildEmptyLogLabel(log, locale)); + // Messages is a bounded WINDOW. Anything older has been sealed into ActivityLogSegment + // satellites under {activityPath}/_Log and is still durable — say so rather than letting the + // pane silently look like the run started midway. (The pane is a 320px scroller: it never + // showed more than ~20 rows at once anyway, and it used to rebuild one row per message on + // EVERY stream emission — so a 5,000-line script rebuilt 5,000 rows 5,000 times.) + var archived = log.TotalMessageCount - log.Messages.Count; + if (archived > 0) + stack = stack.WithView(Controls + .Label(LocalizationCatalog.Get("ui.activityEarlierMessages", locale, archived)) + .WithStyle("font-style: italic; color: var(--neutral-foreground-hint);")); + foreach (var msg in log.Messages) { var (color, tag) = LevelTag(msg.LogLevel); diff --git a/src/MeshWeaver.Graph/Configuration/ActivityLogSegmentNodeType.cs b/src/MeshWeaver.Graph/Configuration/ActivityLogSegmentNodeType.cs new file mode 100644 index 0000000000..151986621d --- /dev/null +++ b/src/MeshWeaver.Graph/Configuration/ActivityLogSegmentNodeType.cs @@ -0,0 +1,58 @@ +using MeshWeaver.Data; +using MeshWeaver.Graph.Security; +using MeshWeaver.Mesh; +using MeshWeaver.Mesh.Security; +using MeshWeaver.Mesh.Services; +using MeshWeaver.Messaging; +using Microsoft.Extensions.DependencyInjection; + +namespace MeshWeaver.Graph.Configuration; + +/// +/// Registers the ActivityLogSegment node type — a sealed slice of an activity's transcript, +/// written once at {activityPath}/_Log/{index:D6} when the messages scroll out of the +/// activity's bounded window (see ). +/// +/// System-generated satellite, exactly like : excluded from search +/// and create contexts, access delegated to the MainNode via SatelliteAccessRule. It carries no +/// views — a segment is read as part of its activity's log, never as a page of its own. +/// +public static class ActivityLogSegmentNodeType +{ + /// The node-type identifier string for activity log segment nodes. + public const string NodeType = ActivityLogAppender.SegmentNodeType; + + /// + /// Registers the node type on the mesh builder: adds the MeshNode definition, excludes it from + /// autocomplete, and wires the satellite access rule that delegates access to the parent (MainNode). + /// + /// The mesh builder type. + /// The mesh builder to configure. + /// The same builder, to allow fluent chaining. + public static TBuilder AddActivityLogSegmentType(this TBuilder builder) where TBuilder : MeshBuilder + { + builder.AddMeshNodes(CreateMeshNode()); + builder.AddAutocompleteExcludedTypes(NodeType); + builder.ConfigureServices(services => + { + services.AddSingleton(sp => + new SatelliteAccessRule(NodeType, sp.GetRequiredService())); + return services; + }); + return builder; + } + + /// + /// Builds the MeshNode definition for the activity log segment node type. + /// + /// The ActivityLogSegment MeshNode definition. + public static MeshNode CreateMeshNode() => new(NodeType) + { + Name = "Activity Log Segment", + IsSatelliteType = true, + ExcludeFromContext = new HashSet { "search", "create" }, + HubConfiguration = config => config + .AddMeshDataSource(source => source + .WithContentType()) + }; +} diff --git a/src/MeshWeaver.Graph/Configuration/GraphConfigurationExtensions.cs b/src/MeshWeaver.Graph/Configuration/GraphConfigurationExtensions.cs index 4820434b71..829d5e2575 100644 --- a/src/MeshWeaver.Graph/Configuration/GraphConfigurationExtensions.cs +++ b/src/MeshWeaver.Graph/Configuration/GraphConfigurationExtensions.cs @@ -61,6 +61,7 @@ public TBuilder AddGraph() .AddTeamsConversationType() .AddGraphSubscriptionType() .AddActivityType() + .AddActivityLogSegmentType() .AddUserActivityType() .AddHomeTabType() .AddHomeConfigType() diff --git a/src/MeshWeaver.Graph/Configuration/NodeTypeCompilationActivity.cs b/src/MeshWeaver.Graph/Configuration/NodeTypeCompilationActivity.cs index 364fc750d1..07ff59213f 100644 --- a/src/MeshWeaver.Graph/Configuration/NodeTypeCompilationActivity.cs +++ b/src/MeshWeaver.Graph/Configuration/NodeTypeCompilationActivity.cs @@ -130,32 +130,7 @@ public static void AppendLogs( IMessageHub hub, string? activityPath, IReadOnlyList messages, ILogger logger) { if (string.IsNullOrEmpty(activityPath) || messages.Count == 0) return; - try - { - // Set the property on the activity's stream — GetMeshNodeStream - // auto-detects own vs remote (the compile-activity handler runs ON - // the activity hub, so this is its OWN stream — GetRemoteStream - // would throw "Owner cannot be the same as the subscriber"). The - // Update rides the synchronization protocol; no message post. - // Impersonate System: an infrastructure activity-log write must not fail closed when the - // calling thread carries no user (see Start). - var accessService = hub.ServiceProvider.GetService(); - using (accessService?.ImpersonateAsSystem()) - hub.GetWorkspace().GetMeshNodeStream(activityPath!) - .Update(current => - current?.Content is ActivityLog log - ? current with { Content = log.Append(messages) } - : current!) - .Subscribe( - _ => { }, - ex => logger.LogDebug(ex, - "Compile-activity AppendLog failed for {Path} (best-effort, ignored)", activityPath)); - } - catch (Exception ex) - { - logger.LogDebug(ex, - "Compile-activity AppendLog threw for {Path} (best-effort, ignored)", activityPath); - } + Write(hub, activityPath, messages, mutate: null, logger, "AppendLog"); } /// @@ -172,33 +147,9 @@ public static void Complete( IReadOnlyList messages, ILogger logger) { if (string.IsNullOrEmpty(activityPath)) return; - try - { - // Impersonate System: terminal activity-log write must not fail closed (see Start). - var accessService = hub.ServiceProvider.GetService(); - using (accessService?.ImpersonateAsSystem()) - hub.GetWorkspace().GetMeshNodeStream(activityPath!) - .Update(current => - current?.Content is ActivityLog log - ? current with - { - Content = log.Append(messages) with - { - Status = status, - End = DateTime.UtcNow, - } - } - : current!) - .Subscribe( - _ => { }, - ex => logger.LogDebug(ex, - "Compile-activity Complete failed for {Path} (best-effort, ignored)", activityPath)); - } - catch (Exception ex) - { - logger.LogDebug(ex, - "Compile-activity Complete threw for {Path} (best-effort, ignored)", activityPath); - } + Write(hub, activityPath, messages, + log => log with { Status = status, End = DateTime.UtcNow }, + logger, "Complete"); } /// @@ -221,47 +172,46 @@ private static void Update( IMessageHub hub, string? activityPath, ActivityStatus status, string? error, ILogger logger) { if (string.IsNullOrEmpty(activityPath)) return; + Write(hub, activityPath, + error is { Length: > 0 } + ? [new LogMessage(error, Microsoft.Extensions.Logging.LogLevel.Error)] + : [], + log => log with { Status = status, End = DateTime.UtcNow }, + logger, "Update"); + } + /// + /// The one write path behind , and + /// : hands the messages + the log change to + /// , which appends through the activity's own stream, + /// keeps Messages bounded, and flushes the overflow into a segment satellite. + /// + /// Impersonate System: a compile-activity write is INFRASTRUCTURE observability — it must not + /// fail closed when the calling thread carries no user (background recompile, grain activation) or a + /// non-writer user (a compile on a read-only partition). Observable.Using holds the scope + /// across the COLD Append's Subscribe; a plain using would have lapsed before the subscribe + /// ran, and the write would post context-null. + /// + private static void Write( + IMessageHub hub, string? activityPath, IReadOnlyList messages, + Func? mutate, ILogger logger, string what) + { try { - // Set the terminal status property on the activity's stream. - // GetMeshNodeStream auto-detects own vs remote — the compile-activity - // handler runs ON the activity hub, so this writes through its OWN - // stream (GetRemoteStream would throw "Owner cannot be the same as - // the subscriber"). The Update rides the synchronization protocol; - // no UpdateNodeRequest message post. - // Impersonate System: a terminal compile-activity write is INFRASTRUCTURE - // observability — it must not fail closed when the calling thread carries no - // user (background recompile, grain activation) or a non-writer user (a compile - // on a read-only partition). MarkSucceeded/MarkFailed route through here, so - // this is the one place the System scope was missing from its Start/AppendLog/ - // Complete siblings (prod 2026-06-18 phantom-activity storm class). var accessService = hub.ServiceProvider.GetService(); - using (accessService?.ImpersonateAsSystem()) - hub.GetWorkspace().GetMeshNodeStream(activityPath!) - .Update(current => - current?.Content is ActivityLog log - ? current with - { - Content = (error is { Length: > 0 } - ? log.Append(new LogMessage(error, - Microsoft.Extensions.Logging.LogLevel.Error)) - : log) with - { - Status = status, - End = DateTime.UtcNow, - } - } - : current!) - .Subscribe( - _ => { }, - ex => logger.LogDebug(ex, - "Compile-activity Update failed for {Path} (best-effort, ignored)", activityPath)); + Observable.Using( + () => accessService?.ImpersonateAsSystem() + ?? System.Reactive.Disposables.Disposable.Empty, + _ => ActivityLogAppender.Append(hub, activityPath!, messages, mutate, logger)) + .Subscribe( + _ => { }, + ex => logger.LogDebug(ex, + "Compile-activity {What} failed for {Path} (best-effort, ignored)", what, activityPath)); } catch (Exception ex) { logger.LogDebug(ex, - "Compile-activity Update threw for {Path} (best-effort, ignored)", activityPath); + "Compile-activity {What} threw for {Path} (best-effort, ignored)", what, activityPath); } } diff --git a/src/MeshWeaver.Kernel.Hub/ActivityLogLogger.cs b/src/MeshWeaver.Kernel.Hub/ActivityLogLogger.cs index e3e4306428..6d14f7b16e 100644 --- a/src/MeshWeaver.Kernel.Hub/ActivityLogLogger.cs +++ b/src/MeshWeaver.Kernel.Hub/ActivityLogLogger.cs @@ -4,6 +4,7 @@ using System.Text.Json; using MeshWeaver.Data; using MeshWeaver.Mesh; +using MeshWeaver.Mesh.Services; using MeshWeaver.Messaging; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; @@ -31,6 +32,21 @@ internal sealed class ActivityLogLogger(IMessageHub hub, string activityLogPath) // under System (Permission.All) — same rule as compile (#2) / user-activity (#3). private readonly AccessService? _accessService = hub.ServiceProvider.GetService(); private ImmutableList _messages = ImmutableList.Empty; + // Incremental severity roll-up, so the published log's terminal status never depends on how much + // of the transcript the head window still holds. + private LogLevel _maxSeverity = LogLevel.Trace; + + // 🚨 Window state — read and written ONLY under _publishLock, alongside the publish itself. + // This logger is the SINGLE writer of its activity node (it re-asserts whole content on every + // flush rather than patching), so it seals its own overflow directly and needs none of + // ActivityLogAppender's claim protocol: there is no second appender to race. + // + // _sealedCount / _sealedSegments advance ONLY in the segment write's success callback. Until + // then the messages stay in the published window, so a failed segment write costs a bigger + // window and a retry on the next flush — never a lost line, and never a watchdog. + private int _sealedCount; + private int _sealedSegments; + private bool _sealInFlight; // 🚨 Terminal-settle state — guarded by _publishLock, together with every // publish. Once Complete has stored a terminal status here, EVERY subsequent @@ -108,6 +124,7 @@ public void Log( lock (_lock) { _messages = _messages.Add(entry); + if (logLevel > _maxSeverity && logLevel != LogLevel.None) _maxSeverity = logLevel; } // Best-effort push, throttled. Failures never surface into the script — @@ -199,17 +216,28 @@ private void PublishSnapshot() private void PublishSnapshotLocked() { ImmutableList snapshot; - lock (_lock) { snapshot = _messages; } + LogLevel maxSeverity; + lock (_lock) { snapshot = _messages; maxSeverity = _maxSeverity; } try { + // 🚨 Seal the overflow BEFORE building the payload. Without this, every 100 ms flush + // re-posts the WHOLE transcript — so a script logging N lines serialises O(N²) bytes onto + // one node, the shape that burned ~719 MB on a single memex-cloud activity. The window + // makes each flush O(1); the sealed lines stay durable in _Log segment satellites. + SealOverflowLocked(snapshot); + var window = snapshot.GetRange(_sealedCount, snapshot.Count - _sealedCount); + // We don't round-trip the MeshNode through a query — we just post a // fresh ActivityLog content payload. The node hub's DataChangeRequest // handler merges it (the Content field is a POCO, so Updates([node]) // replaces Content wholesale — fine for append-only Messages). var log = new ActivityLog("ScriptExecution") { - Messages = snapshot, + Messages = window, + MessageCount = snapshot.Count, + MaxSeverity = maxSeverity, + SegmentCount = _sealedSegments, Status = _terminalStatus ?? ActivityStatus.Running, End = _terminalEnd, ReturnValue = _terminalReturnValue @@ -241,4 +269,70 @@ private void PublishSnapshotLocked() } catch { /* never let logging break the script */ } } + + /// + /// Moves everything above out of the published window + /// and into an satellite. MUST be called under + /// , which is also what makes the "one seal at a time" flag sound. + /// + /// The counters advance in the write's SUCCESS callback, not here — so a failed segment write + /// leaves the lines in the window and the next flush retries the same slice with a deterministic + /// (re-used) segment index. Nothing is lost, nothing is duplicated, and no timer is involved. + /// + private void SealOverflowLocked(ImmutableList snapshot) + { + if (_sealInFlight) return; + var unsealed = snapshot.Count - _sealedCount; + if (unsealed <= ActivityLog.MessageWindowLimit) return; + + var take = unsealed - ActivityLog.MessageWindowKeep; + var firstOrdinal = _sealedCount; + var index = _sealedSegments; + var chunk = snapshot.GetRange(firstOrdinal, take); + + var meshService = hub.ServiceProvider.GetService(); + if (meshService is null) return; + + var segmentId = index.ToString("D6"); + var segmentNode = new MeshNode(segmentId, ActivityLogAppender.SegmentNamespace(activityLogPath)) + { + Name = $"Log {firstOrdinal}-{firstOrdinal + take - 1}", + NodeType = ActivityLogAppender.SegmentNodeType, + // Satellites delegate access to a real main node; the activity's own path carries none. + MainNode = SatelliteTableMapping.OwnerOfSatellitePath(activityLogPath), + State = MeshNodeState.Active, + Content = new ActivityLogSegment + { + Id = segmentId, + FirstOrdinal = firstOrdinal, + ActivityPath = activityLogPath, + Messages = chunk, + }, + }; + + _sealInFlight = true; + // CreateOrUpdateNode (not CreateNode): a retried seal re-writes the SAME index with the same + // content rather than failing on an existing path. + // Observable.Using holds the System scope across the COLD write's Subscribe — a plain `using` + // would have lapsed before the subscribe ran and the write would post context-null. + Observable.Using( + () => _accessService?.ImpersonateAsSystem() ?? Disposable.Empty, + _ => meshService.CreateOrUpdateNode(segmentNode)) + .Subscribe( + _ => + { + lock (_publishLock) + { + _sealedCount = firstOrdinal + take; + _sealedSegments = index + 1; + _sealInFlight = false; + } + }, + _ => + { + // Best-effort, like every other write on this path: the lines stay in the window + // and the next flush retries them. + lock (_publishLock) _sealInFlight = false; + }); + } } diff --git a/src/MeshWeaver.Mcp/McpMeshPlugin.cs b/src/MeshWeaver.Mcp/McpMeshPlugin.cs index 3ae688961f..0e47e62410 100644 --- a/src/MeshWeaver.Mcp/McpMeshPlugin.cs +++ b/src/MeshWeaver.Mcp/McpMeshPlugin.cs @@ -789,7 +789,7 @@ private static string FormatDiagnosticsJson(IReadOnlyList diagno /// Budget for the dispatch acknowledgement (default 120). /// Dispatch-status JSON. [McpServerTool(Title = "Execute a Code node", Destructive = true, Idempotent = false, OpenWorld = false)] - [Description("Runs an executable Code node's C# on the mesh kernel. The target node must have `CodeConfiguration.IsExecutable == true`. Returns as soon as the run has been ACCEPTED, not when it finishes: `{status:'Dispatched', activityPath}` — poll `get @{activityPath}` for the live log messages and the terminal status (Running → Succeeded/Failed), which is also where a script's own exception surfaces. Any dispatch failure (node missing or unreadable, not executable, activity could not be created, request undeliverable) comes back as `{status:'Error', message}` — a 'Dispatched' reply always means a real activity exists. Use to run import/test scripts from MCP without needing a UI click.")] + [Description("Runs an executable Code node's C# on the mesh kernel. The target node must have `CodeConfiguration.IsExecutable == true`. Returns as soon as the run has been ACCEPTED, not when it finishes: `{status:'Dispatched', activityPath}` — poll `get @{activityPath}` for the live log messages and the terminal status (Running → Succeeded/Failed), which is also where a script's own exception surfaces. `Messages` on the activity is the MOST RECENT window (`MessageCount` is the true total); once a run gets long the older lines are archived into segment nodes — list them with `search 'path:{activityPath}/_Log scope:children'` and read one to see that slice. Any dispatch failure (node missing or unreadable, not executable, activity could not be created, request undeliverable) comes back as `{status:'Error', message}` — a 'Dispatched' reply always means a real activity exists. Use to run import/test scripts from MCP without needing a UI click.")] public Task ExecuteScript( [Description("Path to an executable Code node (e.g., @Systemorph/FutuRe/EuropeRe/AcmeSubmission2025/Script/ImportLargeClaims). Must be `IsExecutable=true`.")] string path, [Description("Seconds to wait for the dispatch acknowledgement (NOT for the script to finish). Default 120.")] int timeoutSeconds = 120) diff --git a/src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs b/src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs new file mode 100644 index 0000000000..bbc1379749 --- /dev/null +++ b/src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs @@ -0,0 +1,169 @@ +using System.Reactive.Linq; +using MeshWeaver.Data; +using MeshWeaver.Mesh.Services; +using MeshWeaver.Messaging; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; + +namespace MeshWeaver.Mesh; + +/// +/// The ONE way to write log messages onto a persisted activity MeshNode. Appends through the +/// canonical GetMeshNodeStream(path).Update(...), keeps +/// bounded, and flushes whatever scrolls out of the window into an +/// satellite under {activityPath}/_Log. +/// +/// Why bounding is the whole point. Every stream.Update re-serialises the entire +/// MeshNode.Content to compute its patch, so N appends onto one growing list cost O(N²) in CPU +/// and allocation — measured at ~719 MB of serialisation for a single memex-cloud import activity +/// (5,239 writes over a 141 kB node), and the dominant term in that pod's CFS throttling. A delta +/// field cannot fix it: the cross-hub path ships an RFC 7396 merge patch, which clones a changed array +/// whole, and the three-way merge's base extraction clones the previous array as well. With the head +/// bounded, each write is O(window) = O(1) and the activity as a whole is O(N). +/// +/// Below the window nothing changes. An activity that never exceeds +/// takes exactly the same single write per append it +/// always did, with byte-identical content — so short activities (nearly all of them) are untouched +/// and only the long ones, the ones that actually hurt, take the new path. +/// +/// The flush is lock-free and loses nothing. A seal is CLAIMED inside the head's update +/// lambda (), which the owning hub serialises — so exactly one +/// appender can claim each slice, and no two claims overlap. The claimed messages stay on the head +/// until the segment write succeeds; only then are they trimmed +/// (). A crash or a failed segment write therefore loses nothing +/// and needs no watchdog: the next append re-attempts the same slice, because the claim is still +/// standing and the messages are still there. +/// +/// Stateless static helpers — no DI service, no per-path queue, no lock. Per +/// Doc/Architecture/AsynchronousCalls.md → "Static handlers compose". +/// +public static class ActivityLogAppender +{ + /// The namespace segment holding an activity's sealed log segments. + public const string SegmentNamespaceSegment = "_Log"; + + /// The node type of a sealed log segment satellite. + public const string SegmentNodeType = "ActivityLogSegment"; + + /// The namespace an activity's satellites live in. + /// Path of the activity node. + /// {activityPath}/_Log. + public static string SegmentNamespace(string activityPath) => + $"{activityPath}/{SegmentNamespaceSegment}"; + + /// + /// Appends to the activity at , applying + /// (terminal status, End, ReturnValue, …) in the SAME write so a reader + /// can never observe the terminal status before the lines that explain it. + /// + /// Cold — the write runs on Subscribe. Emits the updated activity node once, then completes. + /// When the append overflows the window the returned observable also covers the segment write and + /// the trim, so a subscriber that waits for completion knows the flush has landed. + /// + /// The hub performing the write. + /// Path of the activity node. + /// Messages to append; may be empty when carries the change. + /// Optional additional change to the log, applied after the append in the same write. + /// Logger for best-effort diagnostics. + /// A cold observable emitting the updated activity node. + public static IObservable Append( + IMessageHub hub, + string activityPath, + IReadOnlyList messages, + Func? mutate = null, + ILogger? logger = null) + { + ArgumentNullException.ThrowIfNull(hub); + ArgumentException.ThrowIfNullOrEmpty(activityPath); + + var workspace = hub.GetWorkspace(); + var options = hub.JsonSerializerOptions; + var stream = workspace.GetMeshNodeStream(activityPath); + + return stream + .Update(node => + { + // ContentAs, never `is ActivityLog`: a cross-hub update lambda diffs against a LOCAL + // mirror whose Content can be a degraded JsonElement — a plain type test would be + // null, the lambda would no-op, and the patch would never be sent. + if (node.ContentAs(options, logger) is not { } log) return node; + var updated = log.Append(messages); + if (mutate is not null) updated = mutate(updated); + // Claim AFTER the append + mutate so the decision sees the final window size. + return node with { Content = updated.ClaimSeal() }; + }) + .SelectMany(updated => FlushClaimedSeal(hub, activityPath, stream, updated, options, logger)); + } + + /// + /// Writes the currently claimed slice into its segment satellite and trims it off the head. A + /// no-op — passing the node straight through — when no seal is claimed, which is the common case. + /// + /// Reads the claim off the node the append just produced rather than off a captured local, so + /// it also picks up a claim left standing by an earlier append whose segment write failed. That is + /// the entire recovery mechanism: no timer, no retry loop, no watchdog. + /// + private static IObservable FlushClaimedSeal( + IMessageHub hub, + string activityPath, + MeshNodeStreamHandle stream, + MeshNode updated, + System.Text.Json.JsonSerializerOptions options, + ILogger? logger) + { + if (updated.ContentAs(options, logger) is not { SealingCount: > 0 } log) + return Observable.Return(updated); + + var sealedMessages = log.SealedMessages; + if (sealedMessages.Count == 0) + return Observable.Return(updated); + + var meshService = hub.ServiceProvider.GetService(); + if (meshService is null) + { + // No mesh service (a bare test hub): the claim stays standing and the messages stay on + // the head — degraded to the pre-window behaviour, never lost. + logger?.LogDebug( + "Activity {Path}: no IMeshService, cannot seal {Count} messages into a segment", + activityPath, sealedMessages.Count); + return Observable.Return(updated); + } + + // The index IS the segment count, so a retried flush re-writes the same path with the same + // content rather than minting a duplicate. CreateOrUpdateNode makes that idempotent. + var index = log.SegmentCount; + var segmentId = index.ToString("D6"); + var segmentNode = new MeshNode(segmentId, SegmentNamespace(activityPath)) + { + Name = $"Log {log.SealedFirstOrdinal}-{log.SealedFirstOrdinal + sealedMessages.Count - 1}", + NodeType = SegmentNodeType, + // A satellite delegates access to the content entity its OWNER delegates to — never to the + // activity node (itself a satellite), whose path carries no permissions of its own. + MainNode = string.IsNullOrEmpty(updated.MainNode) ? activityPath : updated.MainNode, + State = MeshNodeState.Active, + Content = new ActivityLogSegment + { + Id = segmentId, + FirstOrdinal = log.SealedFirstOrdinal, + ActivityPath = activityPath, + Messages = sealedMessages, + }, + }; + + return meshService.CreateOrUpdateNode(segmentNode) + .SelectMany(_ => stream.Update(node => + node.ContentAs(options, logger) is { SealingCount: > 0 } current + ? node with { Content = current.CompleteSeal() } + : node)) + // A failed segment write must never fail the work being logged. The claim stays standing + // and the messages stay on the head, so the next append retries the identical slice. + .Catch(ex => + { + logger?.LogDebug(ex, + "Activity {Path}: sealing {Count} messages into segment {Index} failed; " + + "they stay on the head and the next append retries", + activityPath, sealedMessages.Count, index); + return Observable.Return(updated); + }); + } +} diff --git a/src/MeshWeaver.Mesh.Contract/SatelliteTableMapping.cs b/src/MeshWeaver.Mesh.Contract/SatelliteTableMapping.cs index 739d436f83..b32e56787f 100644 --- a/src/MeshWeaver.Mesh.Contract/SatelliteTableMapping.cs +++ b/src/MeshWeaver.Mesh.Contract/SatelliteTableMapping.cs @@ -25,6 +25,11 @@ public sealed record SatelliteTableMapping(string Segment, string Table, params public static IReadOnlyList Defaults { get; } = [ new SatelliteTableMapping("_Activity", "activities", "Activity"), + // An activity's sealed log slices ({activityPath}/_Log/{index}) — same table as the activity + // they belong to, so a partition's activity data stays in one place and no new table is + // provisioned. The path already contains _Activity, so segment resolution agrees either way; + // the entry is what makes a nodeType-filtered query find them. + new SatelliteTableMapping("_Log", "activities", "ActivityLogSegment"), new SatelliteTableMapping("_UserActivity", "user_activities", "UserActivity"), new SatelliteTableMapping("_Thread", "threads", "Thread", "ThreadComposer"), new SatelliteTableMapping("_ThreadMessage", "threads", "ThreadMessage"), diff --git a/src/MeshWeaver.Messaging.Hub/Localization/strings.de.json b/src/MeshWeaver.Messaging.Hub/Localization/strings.de.json index 7360128c63..63f470df02 100644 --- a/src/MeshWeaver.Messaging.Hub/Localization/strings.de.json +++ b/src/MeshWeaver.Messaging.Hub/Localization/strings.de.json @@ -412,6 +412,7 @@ "ui.sourceChanged": "Quelle geändert — muss kompiliert werden", "ui.upToDate": "Aktuell", "ui.notYetRun": "Noch nicht ausgeführt.", + "ui.activityEarlierMessages": "… {0} frühere Meldungen archiviert", "ui.activityNoOutput": "Dieser Lauf hat keine Ausgabe erzeugt.", "ui.running": "Wird ausgeführt…", "ui.statusDone": "Fertig", diff --git a/src/MeshWeaver.Messaging.Hub/Localization/strings.en.json b/src/MeshWeaver.Messaging.Hub/Localization/strings.en.json index 25159eeeb8..48b01e9d91 100644 --- a/src/MeshWeaver.Messaging.Hub/Localization/strings.en.json +++ b/src/MeshWeaver.Messaging.Hub/Localization/strings.en.json @@ -412,6 +412,7 @@ "ui.sourceChanged": "Source changed — needs compile", "ui.upToDate": "Up to date", "ui.notYetRun": "Not yet run.", + "ui.activityEarlierMessages": "… {0} earlier messages archived", "ui.activityNoOutput": "This run produced no output.", "ui.running": "Running…", "ui.statusDone": "Done", diff --git a/test/MeshWeaver.Graph.Test/StaticRepoImportActivityWriteCountTest.cs b/test/MeshWeaver.Graph.Test/StaticRepoImportActivityWriteCountTest.cs index 80cc8171a9..6ee95e5069 100644 --- a/test/MeshWeaver.Graph.Test/StaticRepoImportActivityWriteCountTest.cs +++ b/test/MeshWeaver.Graph.Test/StaticRepoImportActivityWriteCountTest.cs @@ -2,6 +2,7 @@ using System.Collections.Generic; using System.Linq; using System.Reactive.Linq; +using System.Text.Json; using MeshWeaver.Data; using MeshWeaver.Graph.Configuration; using MeshWeaver.Hosting.Monolith.TestBase; @@ -9,6 +10,7 @@ using MeshWeaver.Mesh; using MeshWeaver.Mesh.Services; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; using Xunit; namespace MeshWeaver.Graph.Test; @@ -66,6 +68,170 @@ public async Task ImportActivityWrites_DoNotGrowWithTheNumberOfItemsInAPhase() "the write count is per-phase, so it must stay well under the per-item count"); } + /// How many messages each side of the head-size comparison appends. + private const int ShortActivity = ActivityLog.MessageWindowLimit; // never overflows the window + private const int LongActivity = ActivityLog.MessageWindowLimit * 4; // overflows it three times over + + /// + /// Messages per append call. Purely a test-runtime concession: a cross-hub activity write costs + /// ~0.2 s here, so 2,500 single-message appends would take longer than the whole suite's budget. + /// It does not soften what is measured — the quantity under test is the SIZE of the content each + /// write serialises, and the window behaves identically however the messages arrive. + /// + private const int AppendBatch = 25; + + /// + /// 🚨 THE SECOND REGRESSION GUARD: the cost of ONE append must not depend on how much the activity + /// has already logged. + /// + /// Batching per phase (the test above) bounds how MANY writes an import makes. It does nothing + /// about how EXPENSIVE each one is: every stream.Update re-serialises the whole + /// MeshNode.Content to compute its patch, so appending to a list that keeps growing is O(N²) + /// however few writes you make — and the cross-hub path is worse, since an RFC 7396 merge patch + /// clones the changed array whole and the three-way merge clones the previous one too, shipping + /// ~2N elements to append one. No delta field escapes that; only bounding the head does. + /// + /// What is asserted is the SERIALISED SIZE of the activity's content — the exact quantity + /// every write pays, measured, not modelled — after a short activity and after one four times as + /// long. They must be within a small constant of each other. Before the window they differed by the + /// message ratio (4×); the absolute bound below is what stops both sides simply growing together. + /// + [Fact(Timeout = 240000)] + public async Task AppendCost_DoesNotGrowWithTheLengthOfTheActivity() + { + var shortRun = await AppendAndMeasure(ShortActivity); + var longRun = await AppendAndMeasure(LongActivity); + + Output.WriteLine( + $"head bytes: {ShortActivity} msgs -> {shortRun.HeadBytes:N0}; {LongActivity} msgs -> {longRun.HeadBytes:N0}" + + $" (ratio {(double)longRun.HeadBytes / shortRun.HeadBytes:F2}×)"); + Output.WriteLine( + $"cumulative serialised bytes: {ShortActivity} msgs -> {shortRun.CumulativeBytes:N0}; " + + $"{LongActivity} msgs -> {longRun.CumulativeBytes:N0}"); + Output.WriteLine( + $"activity writes: {ShortActivity} msgs -> {shortRun.Writes}; {LongActivity} msgs -> {longRun.Writes}; " + + $"segments sealed: {shortRun.Segments} / {longRun.Segments}"); + + longRun.HeadBytes.Should().BeLessThan(shortRun.HeadBytes * 3 / 2, + $"a {LongActivity}-message activity must not carry a bigger head than a {ShortActivity}-message " + + "one — Messages is a bounded window, so the per-append serialisation cost is a constant. " + + "A ratio near 4× means the window is gone and every append is re-serialising the whole " + + "transcript again (the O(n²) shape that burned ~719 MB on one memex-cloud import)."); + + // The write count must stay at one per append call plus the amortised seal traffic — the window + // must not be bought by turning every append into several writes. + var appendCalls = (LongActivity + AppendBatch - 1) / AppendBatch; + longRun.Writes.Should().BeLessThan(appendCalls + longRun.Segments * 2 + 4, + $"{appendCalls} append calls cost one write each; sealing adds ONE extra head write per " + + "segment, amortised over " + + $"{ActivityLog.MessageWindowLimit - ActivityLog.MessageWindowKeep} messages"); + + // …and nothing is lost: every message is either still in the window or durable in a segment. + longRun.TotalMessageCount.Should().Be(LongActivity); + longRun.MessagesInSegments.Should().Be(LongActivity - longRun.MessagesInWindow, + "every message that left the window must be durable in an ActivityLogSegment satellite"); + longRun.Segments.Should().BeGreaterThan(0, "a four-window activity must have sealed segments"); + } + + private sealed record AppendMeasurement( + int HeadBytes, long CumulativeBytes, long Writes, int Segments, + int TotalMessageCount, int MessagesInWindow, int MessagesInSegments); + + /// + /// Creates one activity in a fresh partition, appends log lines to it + /// ONE AT A TIME through the canonical , and reports what each of + /// those appends cost. + /// CumulativeBytes sums the activity content's serialised size at every append — the + /// real work the patcher does per write, observed rather than assumed. + /// + private async Task AppendAndMeasure(int messages) + { + var partition = "Ms" + Guid.NewGuid().ToString("N")[..8]; + var imported = await StaticRepoImporter.ImportSource(Mesh, new FakeRepoSource(partition) + { + Root = new MeshNode(partition) + { + Name = "Append Cost", NodeType = "Space", State = MeshNodeState.Active, + Content = new MarkdownContent { Content = "# Append Cost\n\nfixture." } + }, + Nodes = Pages(partition, 1, "v1") + }) + .FirstAsync().Timeout(120.Seconds()); + imported.Outcome.Should().Be("Imported"); + + var meshService = Mesh.ServiceProvider.GetRequiredService(); + var activityId = "measure" + Guid.NewGuid().ToString("N")[..8]; + var activityPath = $"{partition}/_Activity/{activityId}"; + await meshService.CreateNode(new MeshNode(activityId, $"{partition}/_Activity") + { + Name = "Append cost measurement", + NodeType = ActivityNodeType.NodeType, + MainNode = partition, + State = MeshNodeState.Active, + Content = new ActivityLog(ActivityCategory.DataUpdate) + { + Id = activityId, HubPath = partition, Status = ActivityStatus.Running + } + }).FirstAsync().Timeout(60.Seconds()); + + var cumulative = 0L; + var writes = 0; + MeshNode last = null!; + for (var sent = 0; sent < messages; sent += AppendBatch) + { + var batch = Enumerable.Range(sent, Math.Min(AppendBatch, messages - sent)) + .Select(i => new LogMessage( + $"line {i:D5}: a representative progress line for the measurement", + LogLevel.Information)) + .ToArray(); + last = await ActivityLogAppender.Append(Mesh, activityPath, batch) + .FirstAsync().Timeout(60.Seconds()); + writes++; + cumulative += ContentBytes(last); + } + + var log = last.ContentAs(Mesh.JsonSerializerOptions)!; + var segments = await SegmentsOf(activityPath); + + Output.WriteLine( + $" {messages} messages in {writes} append calls -> node version {last.Version}, " + + $"window {log.Messages.Count}, segments {segments.Count}"); + + return new AppendMeasurement( + HeadBytes: ContentBytes(last), + CumulativeBytes: cumulative, + Writes: last.Version, + Segments: segments.Count, + TotalMessageCount: log.TotalMessageCount, + MessagesInWindow: log.Messages.Count, + MessagesInSegments: segments.Sum(s => s.Messages.Count)); + } + + private int ContentBytes(MeshNode node) => + JsonSerializer.Serialize(node.Content, Mesh.JsonSerializerOptions).Length; + + /// + /// The activity's sealed log segments, listed with a CHILDREN query — never a point-read of a + /// segment path. A point-read of an absent satellite opens the shared stream cache's storm breaker + /// on a path a concurrent write is about to use, and the breaker fast-fails writes too. + /// + private async Task> SegmentsOf(string activityPath) + { + var meshService = Mesh.ServiceProvider.GetRequiredService(); + var change = await meshService + .Query(MeshQueryRequest.FromQuery( + $"path:{ActivityLogAppender.SegmentNamespace(activityPath)} scope:children " + + $"nodeType:{ActivityLogAppender.SegmentNodeType}")) + .Where(c => c.ChangeType == QueryChangeType.Initial) + .FirstAsync().Timeout(30.Seconds()); + return change.Items + .Select(n => n.ContentAs(Mesh.JsonSerializerOptions)) + .Where(s => s is not null) + .Select(s => s!) + .OrderBy(s => s.FirstOrdinal) + .ToArray(); + } + /// /// Runs a two-step import into a fresh partition and returns how many times the SECOND import's /// attempt-activity node was written. From 34489b3b1389f74d70f2b964109f220326d698af Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= Date: Thu, 13 Aug 2026 10:04:57 +0200 Subject: [PATCH 2/2] =?UTF-8?q?docs:=20What's=20New=20=E2=80=94=20long=20a?= =?UTF-8?q?ctivity=20logs=20stay=20fast?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5 (1M context) --- .../2026-08-13-activity-logs-stay-fast.md | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) create mode 100644 src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-13-activity-logs-stay-fast.md diff --git a/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-13-activity-logs-stay-fast.md b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-13-activity-logs-stay-fast.md new file mode 100644 index 0000000000..cf1cdc31e7 --- /dev/null +++ b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-13-activity-logs-stay-fast.md @@ -0,0 +1,17 @@ +--- +Name: Long activity logs stay fast +Category: Feature +Description: A long-running import, sync or script no longer slows down as its log grows — older lines are archived and the activity's cost per line is now constant. +Icon: Sparkle +Order: -20260813 +--- + +# What's New — 13 August 2026 + +## Long activity logs stay fast + +Every progress line an activity wrote used to re-save the entire log. The hundredth line therefore cost a hundred times what the first one did, and a long import, repository sync or script slowed down steadily as it ran — one import on a busy deployment spent hundreds of megabytes of processing on nothing but re-saving its own log, enough to make the whole portal sluggish while it worked. + +An activity now keeps its most recent messages and archives older ones into log segments beside it. The cost of recording a line is the same on line 5,000 as on line 1: in a measured run, quadrupling the number of messages used to cost 15× the processing and now costs 4×. Short activities are completely unchanged. + +Nothing is lost. The activity view shows the recent log with a line telling you how many earlier messages were archived, and the archived segments remain readable in full. Progress indicators, previews and the running-activities strip all still show the true message total.