diff --git a/.claude/skills/mesh-data/SKILL.md b/.claude/skills/mesh-data/SKILL.md index 7854d80560..fb6ced5c22 100644 --- a/.claude/skills/mesh-data/SKILL.md +++ b/.claude/skills/mesh-data/SKILL.md @@ -106,11 +106,13 @@ place without erroring** — the one silent-corruption shape here. Use the ancho (`MeshOperations.EditContent`: exact `oldText` → `newText`, re-verified against live text). A positional splice is legal only carrying a base fingerprint the owner checks. -**Known gap — do not design around it, it is tracked.** `CreateOrUpdateNodeRequest` cannot express a -fold (full-instance mode takes `Content` wholesale; its `Patch` mode is declared but **refused by the -handler**, zero callers). So a caller needing *both* create-if-missing *and* a fold has no route, and -the tempting fallback — decide create-vs-update from a query, then `stream.Update` — is the -eventually-consistent-positive bug in section 2. That is core#4928, blocking core#1174. +**Create-if-missing plus a fold:** use `CreateOrUpdateNodeRequest.WithFolds` (#4980). +The create leg uses the seed verbatim; the update leg applies `Sum`, `KeepExisting`, `Min` or +`Max` to the live stored content. Carry the rule and operand, never a result computed from a +query. `HandleTrackActivity` is the caller example: seed count one, `Sum(1)` thereafter, preserve +first-access time, advance last-access time. This retires #1174's query-dependent create/update +split. Folds retain the existing cross-mirror patch conflict semantics; they do not promise +cluster-wide atomic arithmetic. ### The 3 rules this unifies @@ -319,12 +321,11 @@ hub.GetQuery(id, $"path:{parent} scope:children nodeType:X select:path") // EX .Select(node => node.ContentAs(hub.JsonSerializerOptions)); ``` -The index **trails** the store, so "the index has seen it" implies "the store has it" — the point -read opened on that signal can never be early, and never NotFounds. The same lag that disqualifies a -query for CONTENT is what makes it a safe gate. Creating the node anyway? Skip the check entirely -and use `CreateOrUpdateNodeRequest` — **unless the write is a FOLD** (a counter bump, or anything -computed off the node's current value), which that verb cannot express today (core#4928). There is -no correct shape for create-if-missing + fold yet, so say so rather than falling back to the query. +An index row can still be stale after deletion or failed hydration: "the index has seen it" +is not proof that storage currently holds it. A query can gate discovery, but the subsequent +point read must still handle absence. Creating the node anyway? Skip the check entirely +and use `CreateOrUpdateNodeRequest`. For a counter bump or another live-content fold, attach +`WithFolds` rules to the seed instead of deciding create-vs-update from the query. 🚨 **This is about ONE known path whose value you are GATING on — not about node counts.** The worked counter-example is a token chip that reads `content` out of a diff --git a/src/MeshWeaver.Documentation/Data/Architecture/CqrsAndContentAccess.md b/src/MeshWeaver.Documentation/Data/Architecture/CqrsAndContentAccess.md index b21d8f5b1d..6c7ac201eb 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/CqrsAndContentAccess.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/CqrsAndContentAccess.md @@ -808,7 +808,36 @@ hub.Observe( - The caller doesn't need to check existence — the handler reads persistence and either dispatches `CreateNodeRequest` (when missing) or applies the update branch internally via `stream.Update` (when existing). One audit log. One response shape (`CreateOrUpdateNodeResponse` with `WasCreated`). - **Never delete-then-create.** That pattern races the per-node hub's disposal — a `GetNode` issued shortly after the create returns null because the new request hits the still-tearing-down hub. The upsert handler applies the update via `stream.Update` instead, which routes the merge patch to the live owning hub and keeps `GetNode` consistent. - **Permissions stay specific.** Missing target = `Permission.Create` checked by the inner `CreateNodeRequest`. Existing target = `Permission.Update`, enforced on the patch path by the owning hub's `[RequiresPermission(Update)]` pipeline that `stream.Update` routes to (surfaced to the caller as `UnauthorizedAccessException` by `UpdateRemote`). The upsert request itself declares both via `[CreateOrUpdateNodePermission]` so the routing-layer gate still denies callers that have neither. -- **Patch mode is reserved** for incremental edits (log-line append, view-count bump, status flip): set `request.Patch` to a `Json.Patch.JsonPatch` payload. The handler will apply the patch to the existing node (or to `Node` as the seed when missing) and write the result. (Currently surface-only — patch mode lands when its caller does.) +- **Patch mode is reserved and refused.** A non-null `request.Patch` returns `PatchFailed`; it is not an implemented mutation route. Use `WithFolds` for create-if-missing counters and timestamps, as below. + +### Create-if-missing with a live content fold + +`CreateOrUpdateNodeRequest.WithFolds` carries a rule and operand to the upsert handler. +The create leg takes the supplied seed verbatim; the update leg evaluates the rules against +its live node inside `stream.Update`. No query or caller-side content read decides which leg runs. + +```csharp +var request = new CreateOrUpdateNodeRequest(seed) + .WithFolds(hub.JsonSerializerOptions, folds => folds + .Sum(record => record.AccessCount, 1) + .KeepExisting(record => record.FirstAccessedAt) + .Max(record => record.LastAccessedAt, now)); +``` + +`HandleTrackActivity` uses this shape with a seed count of one. It first checks the user's +partition root in storage so navigation cannot create a partition before onboarding, then issues +one upsert from the stable activity hub with the caller's identity. A stale positive activity +query can no longer select an update whose initial state never arrives (#1174). The tracking +hub's in-flight register still accounts for the detached write until it terminates. + +Folds remove the stale caller read; they do **not** add cluster-wide atomic arithmetic. The +cross-hub write still carries an RFC 7396 patch, with the existing conflict and rebase semantics. +`ActivityTrackingUpsertTest` proves the real handler creates despite a stale positive index, +then preserves the first-access timestamp and increments the stored count on later tracks. +A structured conflict remains `FailureKind = NodeUpsertFailureKind.Conflict` across the response +boundary. This open string vocabulary supplements the unchanged legacy rejection enum; unknown +values remain named failures, and a teardown does not claim that nothing was written. +Activity tracking retains its existing conflict and teardown classification without adding retries. Bulk upserts (e.g. node-tree copy) compose the per-node observable and merge with bounded concurrency so a wide subtree doesn't open every per-node hub simultaneously on the receiving side: diff --git a/src/MeshWeaver.Graph/MeshNodeExtensions.cs b/src/MeshWeaver.Graph/MeshNodeExtensions.cs index 8105647474..d7294f4df2 100644 --- a/src/MeshWeaver.Graph/MeshNodeExtensions.cs +++ b/src/MeshWeaver.Graph/MeshNodeExtensions.cs @@ -231,7 +231,6 @@ private static IMessageDelivery HandleTrackActivity( // The tracking hub is hosted off the mesh ROOT and resolves the shared, // registered mesh-root cache (cache/{meshRootId}). See ActivityTrackingHub. var activityHub = hub.GetActivityTrackingHub(); - var workspace = activityHub.GetWorkspace(); // The tracking hub's own JSON options (WithGraphTypes) know UserActivityRecord — // the caller's hub may not, so use the tracking hub's for typed round-trip. var jsonOptions = activityHub.JsonSerializerOptions; @@ -278,164 +277,82 @@ IObservable AsCaller(Func> work) => "TrackActivity ENTER: userId={UserId} activityPath={Path} type={ActivityType} via activityHub={ActivityHub}", req.UserId, activityPath, req.ActivityType, activityHub.Address); - // workspace.GetMeshNodeStream is backed by the mesh ROOT's shared - // IMeshNodeStreamCache (resolved by DI fallback from the tracking hub) — - // repeat tracks for the same activity path reuse the warm handle. Used ONLY - // to WRITE (stream.Update) the node when it already exists — never to probe an - // absent path (see the GetQuery read below). - var stream = workspace.GetMeshNodeStream(activityPath); - - // First-time creation resolves IMeshService / IStorageAdapter from the mesh ROOT — - // IMeshService is AddScoped, so resolving from a leaf scope would target a hub with - // no CreateNodeRequest handler. The ONBOARD-FIRST gate below probes the user's - // partition root (a read never creates a schema) and skips the write when it's absent. + // Creation and folding are ONE owner-side operation. The query index can retain a + // positive row while its node has no live base; using that row to choose stream.Update + // selected the unsafe update path implicated in #1174. The upsert owns existence + + // hydration, and its + // serialized folds carry rules rather than a count calculated from a stale caller read. + var record = new UserActivityRecord + { + Id = encodedPath, + NodePath = req.NodePath, + UserId = req.UserId, + ActivityType = req.ActivityType, + FirstAccessedAt = now, + LastAccessedAt = now, + AccessCount = 1, + NodeName = req.NodeName, + NodeType = req.NodeType, + Namespace = req.Namespace + }; + var saveNode = MeshNode.FromPath(activityPath) with + { + NodeType = "UserActivity", + Name = req.NodeName ?? encodedPath, + MainNode = req.UserId, + State = MeshNodeState.Active, + Content = record + }; + var request = new CreateOrUpdateNodeRequest(saveNode) + .WithFolds(jsonOptions, folds => folds + .Sum(r => r.AccessCount, 1) + .KeepExisting(r => r.FirstAccessedAt) + .Max(r => r.LastAccessedAt, now)) with + { + RequestedBy = callerCtx?.ObjectId + }; var storage = meshRoot.ServiceProvider.GetService(); - var meshService = meshRoot.ServiceProvider.GetService(); - - // 🚨 Read existence via GetQuery (empty-on-absent), NEVER a point - // GetMeshNodeStream(path).Take(1) probe. On a FIRST-time track the activity node - // does not exist; a point-subscribe to that absent path routes to a RoutingGrain - // NotFound + SYNC_STREAM OnError. Because TrackLogin sits on the COLD-LOGIN hot path - // (every cold page load through UserContextMiddleware.TrackLogin), that failing - // subscribe re-storms the router. A GetQuery over the exact path returns an EMPTY set - // when the node is absent (the documented empty-on-absent behaviour) — no NotFound, no - // resubscribe, nothing to storm — and returns typed Content when present. - // - // The increment is still folded onto the LIVE node inside the owner-serialised Update - // below (FoldOntoLive), so the query's eventual consistency only ever decides - // create-vs-update — never the AccessCount. The create-vs-update race is coalesced by - // the CreateNode catch below, which folds the increment in via stream.Update. - // 🚨 Build (and subscribe) the whole read+write pipeline UNDER the caller's identity. - // WrapWithPerUserRls (SyncedQueryDataSourceExtensions) captures AccessService.Context - // EAGERLY at the GetQuery(...) call — on the workspace's hub (the tracking hub) — so the - // per-user RLS filter must see the caller's identity AT THAT CALL, not only at the write's - // subscribe. RunAs enters the scope BEFORE invoking the work factory, so - // GetQuery is called with the caller's context established on the tracking hub's - // AccessService (fail-closed when callerCtx is null: no context ⇒ empty userId ⇒ the RLS - // wrap yields no rows for the exact-path existence probe, and the subsequent write posts - // context-null and is rejected by PostPipeline). The inner per-write AsCaller calls - // still re-establish the identity on each write's own emission thread — - // AsyncLocal does not flow across Rx scheduler hops, so the outer scope alone is not enough. - var pipeline = AsCaller( - () => workspace - .GetQuery($"UserActivity|{activityPath}", $"path:{activityPath} nodeType:UserActivity select:path,id,namespace,name,nodeType,content") - .Take(1) - .Select(nodes => nodes.FirstOrDefault(n => - string.Equals(n.NodeType, "UserActivity", StringComparison.OrdinalIgnoreCase))) - .SelectMany(existing => + var issuingHub = activityHub.NodeOperationIssuingHub(); + var target = activityHub.NodeOperationTarget(); + + // ONBOARD FIRST: an absent user's root must never be created by navigation/login + // tracking. This authoritative root read creates neither an activity hub nor a schema. + // Re-enter the caller's identity at each Subscribe across the root-probe emission hop. + var rootProbe = AsCaller(() => storage != null + ? storage.Read(req.UserId, jsonOptions).Take(1) + : Observable.Return(null)) + .Catch(probeEx => { - var existingRecord = existing.ContentAs(jsonOptions); - var record = new UserActivityRecord - { - Id = encodedPath, - NodePath = req.NodePath, - UserId = req.UserId, - // Honour the request's ActivityType — Login events from the - // auth middleware fold in here alongside Read events from - // navigation. Same persistence path, different filter axis. - ActivityType = req.ActivityType, - FirstAccessedAt = existingRecord?.FirstAccessedAt ?? now, - LastAccessedAt = now, - AccessCount = (existingRecord?.AccessCount ?? 0) + 1, - NodeName = req.NodeName, - NodeType = req.NodeType, - Namespace = req.Namespace - }; - var saveNode = MeshNode.FromPath(activityPath) with - { - NodeType = "UserActivity", - Name = req.NodeName ?? encodedPath, - MainNode = req.UserId, - State = MeshNodeState.Active, - Content = record - }; - - // 🚨 Fold the increment onto the LIVE node INSIDE the Update lambda, not a - // separately-read snapshot. The owner serializes Updates, so each lambda sees - // the freshest AccessCount and two concurrent tracks can't lose an increment. - MeshNode FoldOntoLive(MeshNode live) - { - var liveRec = live.ContentAs(jsonOptions); - return live with - { - NodeType = "UserActivity", - Name = req.NodeName ?? encodedPath, - MainNode = req.UserId, - State = MeshNodeState.Active, - Content = record with - { - AccessCount = (liveRec?.AccessCount ?? 0) + 1, - FirstAccessedAt = liveRec?.FirstAccessedAt ?? record.FirstAccessedAt, - }, - Version = live.Version, - }; - } - - // Each write runs under the acting user via AsCaller/RunAs: the scope is - // entered at the inner Subscribe (where the write primitive captures - // AccessContext) and left on that same Subscribe rather than on whatever - // thread the write happens to terminate on, - // so cross-hub RLS lets it land. See ActivityRunner for the canonical shape. - if (existing != null) + logger?.LogDebug(probeEx, + "TrackActivity root probe failed for {UserId} — treating as not onboarded.", + req.UserId); + return Observable.Return(null); + }); + var pipeline = rootProbe.SelectMany(userRoot => + { + if (userRoot is null) + { + logger?.LogDebug( + "TrackActivity SKIP create for {Path}: user '{UserId}' has no partition root yet " + + "(not onboarded). Activity tracking must not create a partition ahead of onboarding.", + activityPath, req.UserId); + return Observable.Empty(); + } + + logger?.LogDebug("TrackActivity UPSERT: {Path}", activityPath); + return AsCaller(() => issuingHub.Observe(request, options => { - // 🚨 `record.AccessCount` is the count derived from the EVENTUALLY-CONSISTENT - // query snapshot above — it is NOT the count this write lands. The value that - // is written comes from FoldOntoLive, which re-reads AccessCount off the LIVE - // node inside the owner-serialised Update. Naming it `count=` made two - // concurrent tracks that merely READ the same stale snapshot look like two - // writers racing the same increment (#3001's second hypothesis, which the - // owner's three-way merge in fact already prevents). Say which number it is. - logger?.LogDebug( - "TrackActivity UPDATE: {Path} querySnapshotCount={SnapshotCount} " - + "(the written count is folded off the live node inside the Update)", - activityPath, record.AccessCount); - return AsCaller(() => stream.Update(FoldOntoLive)); - } - - var rootProbe = AsCaller(() => storage != null - ? storage.Read(req.UserId, jsonOptions).Take(1) - : Observable.Return(null)) - .Catch(probeEx => - { - logger?.LogDebug(probeEx, - "TrackActivity root probe failed for {UserId} — treating as not onboarded.", - req.UserId); - return Observable.Return(null); - }); - - return rootProbe.SelectMany(userRoot => + options = options.WithTarget(target); + return callerCtx is null ? options : options.WithAccessContext(callerCtx); + }) + .SelectMany(reply => { - if (userRoot is null) - { - logger?.LogDebug( - "TrackActivity SKIP create for {Path}: user '{UserId}' has no partition root yet " + - "(not onboarded). Activity tracking must not create a partition ahead of onboarding.", - activityPath, req.UserId); - return Observable.Empty(); - } - - if (meshService != null) - { - logger?.LogDebug("TrackActivity CREATE: {Path}", activityPath); - return AsCaller(() => meshService.CreateNode(saveNode)) - // Race coalesce: a concurrent track for the same path beat us to - // CreateNode — fold our increment in via Update instead of throwing. - .Catch(ex => - { - if (!IsAlreadyExistsRace(ex)) - return Observable.Throw(ex); - logger?.LogDebug( - "TrackActivity CREATE to UPDATE race for {Path}: another concurrent track won; folding via Update.", - activityPath); - return AsCaller(() => stream.Update(FoldOntoLive)); - }); - } - - return storage != null - ? AsCaller(() => storage.Write(saveNode, jsonOptions)) - : Observable.Empty(); - }); - })); + var result = reply.Message; + if (result.Success && result.Node is not null) + return Observable.Return(result.Node); + return Observable.Throw(ActivityUpsertFailure(result, activityPath)); + })); + }); // The write is DETACHED from this request on purpose — TrackLogin sits on the cold-login // hot path and must not stall behind a node write — so the delivery is Processed below @@ -494,8 +411,8 @@ MeshNode FoldOntoLive(MeshNode live) // // 🚨 The cost/value trade-off, stated because a level change owes one. What is // lost is one AccessCount increment and a sub-second LastAccessedAt — a counter - // that FoldOntoLive re-folds off the live node on the very next activity, so it - // self-corrects. What Error cost instead: this line is the fingerprint's outer + // whose next activity folds onto the live node again; a missed increment + // remains lost. What Error cost instead: this line is the fingerprint's outer // text, so every lost race minted a production incident and reopened an // unrelated fixed ticket (#1910) — an Error nobody can act on, about a write the // platform deliberately retried and deliberately bounded. Debug keeps it @@ -522,15 +439,25 @@ MeshNode FoldOntoLive(MeshNode live) return delivery.Processed(); } - /// - /// True for the specific "Node already exists" signal raised by - /// MeshService.CreateNode when persistence rejects a duplicate - /// path. Distinguishes the concurrent-track race from genuine - /// s (validation failures, - /// missing parent, etc.) which must still surface as errors. - /// - private static bool IsAlreadyExistsRace(InvalidOperationException ex) - => ex.Message.StartsWith("Node already exists:", StringComparison.Ordinal); + /// Preserves structured upsert refusals for activity's existing failure policy. + internal static Exception ActivityUpsertFailure(CreateOrUpdateNodeResponse response, string path) + { + if (response.FailureKind == NodeUpsertFailureKind.Conflict) + return new MeshNodeStreamException( + new MeshNodeError(MeshNodeErrorCode.Conflict, path, response.Error ?? "Node upsert failed")); + if (response.FailureKind == NodeUpsertFailureKind.HubTeardown) + return new HubDisposingException(new Address(path), response.Error ?? "Node upsert failed"); + if (response.FailureKind is { } unknown) + return new InvalidOperationException($"Node upsert failed ({unknown}): {response.Error}"); + return response.RejectionReason switch + { + NodeUpsertRejectionReason.AddressRecycling => + new HubDisposingException(new Address(path), response.Error ?? "Node upsert failed"), + NodeUpsertRejectionReason.Unauthorized or NodeUpsertRejectionReason.ValidationFailed => + new UnauthorizedAccessException(response.Error ?? "Access denied"), + _ => new InvalidOperationException(response.Error ?? "Node upsert failed") + }; + } /// /// True when the failure carries the owner's NACK at diff --git a/src/MeshWeaver.Mesh.Contract/CreateNodeRequest.cs b/src/MeshWeaver.Mesh.Contract/CreateNodeRequest.cs index 970e34ea35..12913b56fd 100644 --- a/src/MeshWeaver.Mesh.Contract/CreateNodeRequest.cs +++ b/src/MeshWeaver.Mesh.Contract/CreateNodeRequest.cs @@ -738,6 +738,14 @@ public record CreateOrUpdateNodeResponse(MeshNode? Node) /// Rejection reason if the upsert failed. public NodeUpsertRejectionReason? RejectionReason { get; init; } + /// + /// Open failure classification, when an inner write supplies a more specific outcome. + /// Values such as supplement the legacy + /// without widening its wire enum. Unrecognised values remain + /// named failures; they must not be interpreted as a retry instruction or a known outcome. + /// + public string? FailureKind { get; init; } + /// Success response for a newly-created node. public static CreateOrUpdateNodeResponse Created(MeshNode node, ActivityLog? log = null) => new(node) { WasCreated = true, Log = log }; @@ -785,6 +793,17 @@ public enum NodeUpsertRejectionReason /// framework's to close separately. /// AddressRecycling, + +} + +/// Open vocabulary for an upsert's inner-write failure. Modules may supply other values. +public static class NodeUpsertFailureKind +{ + /// The owner's conflict verdict after its bounded rebase; this adds no retry. + public const string Conflict = "Conflict"; + + /// The operation ended during teardown; whether the write was applied is unknown. + public const string HubTeardown = "HubTeardown"; } /// @@ -809,6 +828,25 @@ public static class NodeUpsertRejection NodeUpsertRejectionReason.AddressRecycling, _ => NodeUpsertRejectionReason.Unknown, }; + + /// + /// Preserves a structured inner-write outcome beside the unchanged legacy reason. + /// A known legacy refusal takes precedence; an unclassified fault supplies no invented kind. + /// + /// The inner write's fault, or null. + public static string? ClassifyFailureKind(Exception? error) + { + if (Classify(error) != NodeUpsertRejectionReason.Unknown) + return null; + if (HubDisposedBeforeResponseException.IsHubDisposedBeforeResponse(error) + || HubDisposingException.IsHubDisposal(error) + || HubDisposingException.IsDisposedContainer(error)) + return NodeUpsertFailureKind.HubTeardown; + return ExceptionChain.Contains(error, e => + e is MeshNodeStreamException { Error.Code: MeshNodeErrorCode.Conflict }) + ? NodeUpsertFailureKind.Conflict + : null; + } } /// diff --git a/src/MeshWeaver.Mesh.Contract/MeshExtensions.cs b/src/MeshWeaver.Mesh.Contract/MeshExtensions.cs index 14e57136da..164b47f6dd 100644 --- a/src/MeshWeaver.Mesh.Contract/MeshExtensions.cs +++ b/src/MeshWeaver.Mesh.Contract/MeshExtensions.cs @@ -5985,7 +5985,7 @@ private static IMessageDelivery HandleCreateOrUpdateNodeRequest( PostFail( LocalizableText.Keyed($"Persistence read failed: {ex.Message}", "activity.node.upsert.readFailed", ("error", ex.Message)), - NodeUpsertRejectionReason.Unknown); + NodeUpsertRejectionReason.Unknown, NodeUpsertRejection.ClassifyFailureKind(ex)); }, () => { @@ -6153,7 +6153,7 @@ void DispatchInnerCreate() PostFail( LocalizableText.Keyed($"Inner CreateNode faulted: {ex.Message}", "activity.node.upsert.innerCreateFaulted", ("error", ex.Message)), - NodeUpsertRejectionReason.Unknown); + NodeUpsertRejectionReason.Unknown, NodeUpsertRejection.ClassifyFailureKind(ex)); }, () => { @@ -6491,7 +6491,7 @@ MeshNode FoldOntoLive(MeshNode mergedNode, MeshNode liveNode) => ("path", node.Path), ("error", ex.Message)) : LocalizableText.Keyed($"Inner UpdateNode faulted: {ex.Message}", "activity.node.upsert.innerUpdateFaulted", ("error", ex.Message)), - reason); + reason, NodeUpsertRejection.ClassifyFailureKind(ex)); return; } hub.NoteRequestStage(request.Id, "UPSERT_UPDATE_COMPLETED_EMPTY"); @@ -6545,7 +6545,7 @@ void PostOk(MeshNode result, bool isCreate, string logLine, string logKey) // would translate a value code reads and logs — the same reason wire identifiers are never // translated — while giving no viewer a German sentence. The LOCALIZED surface is the // ActivityLog on the same response, which is what a viewer actually reads. - void PostFail(LocalizableText refusal, NodeUpsertRejectionReason reason) + void PostFail(LocalizableText refusal, NodeUpsertRejectionReason reason, string? failureKind = null) { hub.NoteRequestStage(request.Id, $"UPSERT_REPLY fail reason={reason}"); var failLog = baseActivity.Append( @@ -6555,7 +6555,10 @@ void PostFail(LocalizableText refusal, NodeUpsertRejectionReason reason) Status = ActivityStatus.Failed, }; hub.Post( - CreateOrUpdateNodeResponse.Fail(refusal.English, reason, failLog), + CreateOrUpdateNodeResponse.Fail(refusal.English, reason, failLog) with + { + FailureKind = failureKind + }, o => o.ResponseFor(request)); } diff --git a/test/MeshWeaver.Graph.Test/ActivityConflictSeverityTest.cs b/test/MeshWeaver.Graph.Test/ActivityConflictSeverityTest.cs index 930de38bca..e76bfa3926 100644 --- a/test/MeshWeaver.Graph.Test/ActivityConflictSeverityTest.cs +++ b/test/MeshWeaver.Graph.Test/ActivityConflictSeverityTest.cs @@ -1,4 +1,6 @@ +using System.Text.Json; using MeshWeaver.Mesh; +using MeshWeaver.Messaging; using Xunit; namespace MeshWeaver.Graph.Test; @@ -72,6 +74,81 @@ public void EveryOtherOwnerVerdict_StaysLoud(MeshNodeErrorCode code) => MeshNodeExtensions.IsConcurrentWriteConflict(Nack(code, "refused")) .Should().BeFalse(); + [Theory] + [InlineData(MeshNodeErrorCode.Conflict, true)] + [InlineData(MeshNodeErrorCode.AccessDenied, false)] + [InlineData(MeshNodeErrorCode.Validation, false)] + [InlineData(MeshNodeErrorCode.NotFound, false)] + [InlineData(MeshNodeErrorCode.OwnerUnreachable, false)] + [InlineData(MeshNodeErrorCode.Unknown, false)] + public void UpsertWireResponse_PreservesTheExistingConflictPolicy(MeshNodeErrorCode code, bool conflict) + { + var failure = new InvalidOperationException("wrapped", new AggregateException(Nack(code, Total))); + var reason = NodeUpsertRejection.Classify(failure); + reason.Should().Be(NodeUpsertRejectionReason.Unknown); + var kind = NodeUpsertRejection.ClassifyFailureKind(failure); + kind.Should().Be(conflict ? NodeUpsertFailureKind.Conflict : null); + var json = JsonSerializer.Serialize(CreateOrUpdateNodeResponse.Fail(Total, reason) with + { + FailureKind = kind + }); + var reply = JsonSerializer.Deserialize(json)!; + var translated = MeshNodeExtensions.ActivityUpsertFailure(reply, "rbuergi/_UserActivity/rbuergi"); + MeshNodeExtensions.IsConcurrentWriteConflict(translated).Should().Be(conflict); + } + + [Fact] + public void AStructuredRecyclingRefusal_RemainsAnActivityTeardown() + { + var reply = CreateOrUpdateNodeResponse.Fail("recycling", NodeUpsertRejectionReason.AddressRecycling); + HubDisposingException.IsHubDisposal(MeshNodeExtensions.ActivityUpsertFailure(reply, "rbuergi/activity")) + .Should().BeTrue(); + HubDisposingException.IsHubDisposal(MeshNodeExtensions.ActivityUpsertFailure( + CreateOrUpdateNodeResponse.Fail("recycling", NodeUpsertRejectionReason.Unknown), "rbuergi/activity")) + .Should().BeFalse("text alone is never evidence of a teardown"); + } + + [Theory] + [InlineData(0)] + [InlineData(1)] + [InlineData(2)] + public void InnerTeardown_RetainsItsPolicyThroughTheUpsertResponse(int shape) + { + var address = new Address("rbuergi/activity"); + Exception inner = shape switch + { + 0 => new HubDisposingException(address, "activity write"), + 1 => new HubDisposedBeforeResponseException("activity", address, "update", address.ToString()), + _ => new ObjectDisposedException("Autofac.LifetimeScope", "nested lifetimes cannot be created") + }; + var failure = new AggregateException(inner); + var reason = NodeUpsertRejection.Classify(failure); + reason.Should().Be(NodeUpsertRejectionReason.Unknown); + var kind = NodeUpsertRejection.ClassifyFailureKind(failure); + kind.Should().Be(NodeUpsertFailureKind.HubTeardown, + "a lost response during teardown is distinct from an intake refusal that applied nothing"); + var reply = JsonSerializer.Deserialize(JsonSerializer.Serialize( + CreateOrUpdateNodeResponse.Fail(inner.Message, reason) with { FailureKind = kind }))!; + HubDisposingException.IsHubDisposal(MeshNodeExtensions.ActivityUpsertFailure(reply, address.ToString())) + .Should().BeTrue(); + NodeUpsertRejection.Classify(new ObjectDisposedException("unrelated resource")) + .Should().Be(NodeUpsertRejectionReason.Unknown); + NodeUpsertRejection.ClassifyFailureKind(new ObjectDisposedException("unrelated resource")) + .Should().BeNull(); + } + + [Fact] + public void AnUnrecognisedFailureKindSurvivesTheWireAndStaysLoud() + { + var reply = JsonSerializer.Deserialize(JsonSerializer.Serialize( + CreateOrUpdateNodeResponse.Fail("module refusal") with { FailureKind = "Module.CustomOutcome" }))!; + reply.FailureKind.Should().Be("Module.CustomOutcome"); + var error = MeshNodeExtensions.ActivityUpsertFailure(reply, "rbuergi/activity"); + error.Message.Should().Contain("Module.CustomOutcome"); + MeshNodeExtensions.IsConcurrentWriteConflict(error).Should().BeFalse(); + HubDisposingException.IsHubDisposal(error).Should().BeFalse(); + } + [Fact] public void AnUnrelatedFault_StaysLoud() // Nothing about the word "conflict" is load-bearing: a plain exception whose text mentions diff --git a/test/MeshWeaver.Graph.Test/ActivityTrackingUpsertTest.cs b/test/MeshWeaver.Graph.Test/ActivityTrackingUpsertTest.cs new file mode 100644 index 0000000000..7a602177fa --- /dev/null +++ b/test/MeshWeaver.Graph.Test/ActivityTrackingUpsertTest.cs @@ -0,0 +1,161 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Reactive.Linq; +using System.Text.Json; +using System.Threading.Tasks; +using MeshWeaver.Data; +using MeshWeaver.Graph.Configuration; +using MeshWeaver.Hosting.Monolith.TestBase; +using MeshWeaver.Mesh; +using MeshWeaver.Mesh.Activity; +using MeshWeaver.Mesh.Security; +using MeshWeaver.Mesh.Services; +using MeshWeaver.Messaging; +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace MeshWeaver.Graph.Test; + +/// +/// The real tracking handler must create and fold independently of an eventually consistent +/// query index (#1174). The test provider models a stale positive at the backend extension seam; +/// messaging, tracking, storage, permissions and the upsert owner are the real mesh. +/// +public class ActivityTrackingUpsertTest(ITestOutputHelper output) : MonolithMeshTestBase(output) +{ + private const string User = "activity-upsert-user"; + private const string SourcePath = User + "/page"; + private const string ActivityPath = User + "/_UserActivity/activity-upsert-user_page"; + + protected override MeshBuilder ConfigureMesh(MeshBuilder builder) => base.ConfigureMesh(builder) + .ConfigureServices(services => services.AddSingleton(new StaleActivityProvider())); + + /// A positive index row is neither a live update base nor evidence of a stored node. + [Fact(Timeout = 120_000)] + public async Task StalePositiveQuery_DoesNotPreventCreate_AndLaterTracksFoldTheLiveCount() + { + var ct = TestContext.Current.CancellationToken; + var identity = new AccessContext { ObjectId = User, Name = User }; + Mesh.ServiceProvider.GetRequiredService().SetHostIdentity(identity); + await NodeFactory.CreateNode(new MeshNode(User) + { + Name = User, NodeType = "User", State = MeshNodeState.Active, + }).Should().Emit(cancellationToken: ct); + + var hub = Mesh.GetActivityTrackingHub(); + var indexed = await hub.ServiceProvider.GetRequiredService() + .RunAs(identity, () => hub.GetWorkspace().GetQuery($"UserActivity|{ActivityPath}", + $"path:{ActivityPath} nodeType:UserActivity select:path,id,namespace,name,nodeType,content")) + .Where(nodes => nodes.Any(n => n.Path == ActivityPath)) + .Take(1).Should().Within(TimeSpan.FromSeconds(15)).Emit(cancellationToken: ct); + indexed.Should().Contain(n => n.Path == ActivityPath, + "the control must prove that the query really reports this absent activity"); + var absent = await Mesh.ServiceProvider.GetRequiredService() + .Read(ActivityPath, Mesh.JsonSerializerOptions) + .Should().Emit(cancellationToken: ct); + absent.Should().BeNull("the index row must be stale, not a real node"); + + await Track(hub, identity, SourcePath); + var first = await ReadActivity(); + first.AccessCount.Should().Be(1, "the create leg must use its seed, not the index's count 999"); + + await Track(hub, identity, SourcePath); + await Track(hub, identity, SourcePath); + var third = await ReadActivity(); + third.AccessCount.Should().Be(3, "each later track must fold onto the live stored count"); + third.FirstAccessedAt.Should().Be(first.FirstAccessedAt); + third.LastAccessedAt.Should().BeOnOrAfter(first.LastAccessedAt); + third.NodePath.Should().Be(SourcePath); + third.UserId.Should().Be(User); + } + + /// Navigation must not create a user's partition ahead of onboarding. + [Fact(Timeout = 60_000)] + public async Task MissingUserRoot_SettlesWithoutCreatingActivity() + { + var ct = TestContext.Current.CancellationToken; + var identity = new AccessContext { ObjectId = User, Name = User }; + Mesh.ServiceProvider.GetRequiredService().SetHostIdentity(identity); + var hub = Mesh.GetActivityTrackingHub(); + await Track(hub, identity, SourcePath); + var storage = Mesh.ServiceProvider.GetRequiredService(); + var root = await storage.Read(User, Mesh.JsonSerializerOptions).Should().Emit(cancellationToken: ct); + var activity = await storage.Read(ActivityPath, Mesh.JsonSerializerOptions).Should().Emit(cancellationToken: ct); + root.Should().BeNull("tracking must leave onboarding in charge of creating the user"); + activity.Should().BeNull("no activity may create an unonboarded user's partition"); + } + + private static async Task Track(IMessageHub hub, AccessContext identity, string sourcePath) + { + var settled = hub.ServiceProvider.GetRequiredService() + .WhenSettled(ActivityPath).Replay(1); + using var subscription = settled.Connect(); + hub.Post(new TrackActivityRequest(sourcePath, User, "Activity probe", "Markdown", User), + options => options.WithAccessContext(identity)); + await settled.Should().Within(TimeSpan.FromSeconds(45)).Emit( + "the detached write must terminate before storage is inspected", + cancellationToken: TestContext.Current.CancellationToken); + } + + private async Task ReadActivity() + { + var node = await Mesh.ServiceProvider.GetRequiredService() + .Read(ActivityPath, Mesh.JsonSerializerOptions) + .Should().Emit(cancellationToken: TestContext.Current.CancellationToken); + node.Should().NotBeNull("tracking must persist despite the stale positive query result"); + return node!.ContentAs(Mesh.GetActivityTrackingHub().JsonSerializerOptions)!; + } + + /// The actual hub serializer preserves open outcomes and the legacy reason. + [Fact] + public void UpsertFailureKinds_RoundTripThroughTheHubSerializer() + { + var options = Mesh.GetActivityTrackingHub().JsonSerializerOptions; + foreach (var kind in new[] + { NodeUpsertFailureKind.Conflict, NodeUpsertFailureKind.HubTeardown, "Module.CustomOutcome" }) + { + var response = CreateOrUpdateNodeResponse.Fail("owner refusal") with { FailureKind = kind }; + var json = JsonSerializer.Serialize(response, options); + var restored = JsonSerializer.Deserialize(json, options)!; + restored.FailureKind.Should().Be(kind); + restored.RejectionReason.Should().Be(NodeUpsertRejectionReason.Unknown, + "the additive vocabulary must not change what a legacy consumer reads"); + } + } + + private sealed class StaleActivityProvider : IMeshQueryProvider + { + public bool Matches(IReadOnlyList queryNamespaces) => queryNamespaces.Contains(User); + + public IObservable> Query(MeshQueryRequest request, JsonSerializerOptions options) + { + var stale = MeshNode.FromPath(ActivityPath) with + { + NodeType = "UserActivity", Name = "Stale activity", MainNode = User, + State = MeshNodeState.Active, + Content = new UserActivityRecord + { + Id = "activity-upsert-user_page", UserId = User, NodePath = SourcePath, + AccessCount = 999, + }, + }; + var matches = request.EffectiveQueries.Any(q => q.Contains(ActivityPath, StringComparison.Ordinal)); + return Observable.Return(new QueryResultChange + { + ChangeType = QueryChangeType.Initial, + Items = matches && stale is T typed ? [typed] : Array.Empty(), + Timestamp = DateTimeOffset.UtcNow, + }); + } + + public IObservable> Autocomplete( + string basePath, string prefix, JsonSerializerOptions options, + AutocompleteMode mode = AutocompleteMode.RelevanceFirst, int limit = 10, + string? contextPath = null, string? context = null) + => Observable.Return((IReadOnlyCollection)Array.Empty()); + + public IObservable Select(string path, string property, JsonSerializerOptions options) + => Observable.Return(default); + } +} diff --git a/test/MeshWeaver.Hosting.Test/ANodeTypesSourcesWaitForItsBundleTest.cs b/test/MeshWeaver.Hosting.Test/ANodeTypesSourcesWaitForItsBundleTest.cs index 4d080aed19..12c3205bd1 100644 --- a/test/MeshWeaver.Hosting.Test/ANodeTypesSourcesWaitForItsBundleTest.cs +++ b/test/MeshWeaver.Hosting.Test/ANodeTypesSourcesWaitForItsBundleTest.cs @@ -131,7 +131,15 @@ public async Task AnAdoptedTypesSources_WaitForABundleThatCarriesThem() await Sync.ReimportAtCommit(Space, CommitA, UserId) .Timeout(TestTimeouts.CrossSilo).Await(cancellationToken); - var adoptedFingerprint = await FingerprintWhen(d => d.CurrentSourceFingerprint is { Length: > 0 }, + // Importing activates the type and can already dispatch its first compile. A source + // fingerprint does not mean that compile has finished: seeding at that point races its + // terminal stamp, which honestly changes provenance back to Compiled. Establish the + // completed initial build before replacing it with the adoption this test exercises. + var adoptedFingerprint = await FingerprintWhen(d => + d.CurrentSourceFingerprint is { Length: > 0 } + && d.CompilationStatus is CompilationStatus.Ok + && d.BuildProvenance is BuildProvenance.Compiled + && !d.IsDirty, cancellationToken); Output.WriteLine($"live fingerprint at {CommitA[..8]}: {adoptedFingerprint}"); StageBundle("widget-a.zip", adoptedFingerprint);