Plugin install: a bulk write the store REFUSED is no longer reported as written (#2361)

src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-26-install-refused-write.md

@@ -0,0 +1,21 @@
+---
+Name: Plugin installs no longer silently drop a node
+Category: Fix
+Description: A package install now re-applies any node the store refused, instead of reporting it as installed while the old content stayed in place.
+Icon: Sparkle
+Order: -20260826
+---
+
+# Plugin installs no longer silently drop a node
+
+Installing a plugin writes its nodes in batches, and each batch is classified as "new" from a
+snapshot taken at the start of the install. If something else wrote one of those paths in the
+meantime, the storage layer correctly refused the batch's write for that node — but the install
+counted it as written anyway, so the partition kept the older content while the install reported
+success. Nothing surfaced it: the node only got its real content on the *next* install of the same
+package.
+
+The installer now notices the refusal and re-applies the node the same way it applies any node that
+already exists — honouring a node you have claimed as your own, skipping one that is already
+identical, and otherwise landing the package's content. So a plugin's content is what the package
+says after one install, and re-installing an unchanged package really does nothing.

src/MeshWeaver.Hosting/Persistence/MonotonicWriteGuardStorageAdapter.cs

@@ -190,15 +190,27 @@ public IObservable<MeshNode> ReadMany(IReadOnlyCollection<string> paths, JsonSer
         var path = latest.Path;
         var resolution = MeshNodeConflictMerge.Merge(latest, stale, options);
 
+        // 🚨 The durable row is DESCRIBED, not just counted. This message ends with "find the
+        // writer", and until #2361 it withheld the only evidence that would identify one: on the
+        // plugins gate's intermittent Skill failure the line named the members that diverged
+        // (Category, Order, Content.instructions) and nothing about WHOSE row had won, so the
+        // writer had to be inferred from repo content and stayed unidentified across three full
+        // local repro attempts. LastModifiedBy is the principal that wrote it; the type/name/
+        // category say which SHAPE of node it is. All cheap scalars — no content is serialized
+        // here (a NodeType's content can be a collectible-ALC type, which reflecting over during
+        // teardown is its own crash class).
         logger?.LogWarning(
             "[MonotonicWriteGuard] CONFLICT on {Path}: a write at Version={IncomingVersion} lost the race against "
             + "the durable Version={StoredVersion}. MeshNode.Version is the owner's monotonic persistence clock "
             + "(MeshNode.NextVersion floors every mint at current+1), so the losing write is a STALE snapshot, not a "
             + "newer state. Resolved by merging into the durable row: merged={MergedMembers}; "
-            + "latest-wins (stale values DROPPED)={OverwrittenMembers}. Find the writer that adopted a stale "
+            + "latest-wins (stale values DROPPED)={OverwrittenMembers}. The durable row: nodeType={StoredNodeType} "
+            + "name='{StoredName}' category='{StoredCategory}' lastModifiedBy='{StoredLastModifiedBy}' "
+            + "lastModified={StoredLastModified:O}. Find the writer that adopted a stale "
             + "own-node snapshot; do not relax this guard.",
             path, stale.Version, latest.Version,
-            Describe(resolution.MergedMembers), Describe(resolution.OverwrittenMembers));
+            Describe(resolution.MergedMembers), Describe(resolution.OverwrittenMembers),
+            latest.NodeType, latest.Name, latest.Category, latest.LastModifiedBy, latest.LastModified);
 
         RecordConflictActivity(path, stale.Version, latest.Version, resolution, options);
 

src/MeshWeaver.PluginCatalog/PackageInstaller.cs

@@ -2261,11 +2261,14 @@ int Order(MeshNode n)
                     LastModified = now,
                 })
                 .ToArray();
-            return Observable.Using(
-                    () => accessService?.ImpersonateAsSystem()
-                          ?? System.Reactive.Disposables.Disposable.Empty,
-                    _ => persistence.WriteManyAndPublishCreated(stamped, options, changeFeed))
-                .Select(written =>
+            // RunAsSystem, never Observable.Using(ImpersonateAsSystem) (#1790): the scope is an
+            // AsyncLocal store/restore pair and Rx can dispose it on a different thread than the
+            // one that created it, latching the subscriber as `system`.
+            // ImpersonationScopeThreadAffinityTest ratchets this; the rest of this file already
+            // uses RunAsSystem for exactly that reason.
+            return accessService.RunAsSystem(
+                    () => persistence.WriteManyAndPublishCreated(stamped, options, changeFeed))
+                .SelectMany(written =>
                 {
                     if (written.Count != batch.Count)
                     {
@@ -2276,10 +2279,77 @@ int Order(MeshNode n)
                             $"Bulk install of '{manifest.Id}' persisted {written.Count}/{batch.Count} "
                             + $"node(s) — storage did not accept: {string.Join(", ", missing)}");
                     }
-                    return (IList<(string, bool)>)batch.Select(n => (n.Path, true)).ToList();
+                    return ReapplyRefused(stamped, written)
+                        .Select(refusedOutcome => (IList<(string, bool)>)batch
+                            .Select(n => (n.Path,
+                                refusedOutcome.TryGetValue(n.Path, out var wrote) ? wrote : true))
+                            .ToList());
                 });
         }
 
+        // 🚨 ACCEPTED IS NOT APPLIED — a bulk emission is not proof the node was written (#2361).
+        //
+        // The bulk path is chosen for nodes the install's ONE bulk read (ReadCurrent, taken before
+        // the root write, the visibility barrier and the access phase) reported ABSENT, and it
+        // writes them STRAIGHT to the storage adapter at the version the repo file carries — 0,
+        // because a node file has no version. That classification is a snapshot from hundreds of
+        // milliseconds earlier: if the path exists by the time the batch lands, version 0 is a
+        // BACKWARD write, and the write-integrity chain does exactly what it must — the store's
+        // version-conditional upsert refuses it, MonotonicWriteGuard merges latest-wins, and the
+        // package's authored values are DROPPED. Both halves report the refusal the same way:
+        // by emitting the DURABLE node instead of the one handed in (IStorageAdapter.Write's
+        // contract, and the guard's "a version ABOVE the one we handed it" rule).
+        //
+        // The count therefore still matches, and reporting `Wrote = true` off the count alone is
+        // the installer asserting a write that never happened: the partition keeps the foreign
+        // content, the install logs success, and the ONLY thing that notices is the plugin gate's
+        // idempotence pin on the NEXT install — where the node now exists, takes the request path,
+        // mints a forward version and finally lands ("re-install of the unchanged snapshot wrote
+        // 2 node(s): Skill/presentation, Skill/slide", intermittently, MeshWeaver#2361).
+        //
+        // So a refused node is re-decided here as exactly what it turned out to be — an EXISTING
+        // node — through DecideAndWrite, the same function the request path uses, against the
+        // durable row the store just handed back. That keeps all three rules the bulk
+        // classification skipped on the (false) grounds that the node was new: the CLAIM fence (a
+        // node the user took off the repo is still not overwritten), the unchanged-skip (a durable
+        // row that already equals the authored node is not rewritten just because its version is
+        // higher), and — for everything else — the validating request path, whose owning hub mints
+        // a FORWARD version so the write actually lands. Never a retry of the bulk write itself:
+        // repeating a version-0 write against a newer row loses again, by construction.
+        //
+        // The outcome per refused path is returned rather than assumed, so `Written`/`WrittenPaths`
+        // report what happened instead of what was attempted.
+        IObservable<IReadOnlyDictionary<string, bool>> ReapplyRefused(
+            IReadOnlyList<MeshNode> requested, IReadOnlyList<MeshNode> written)
+        {
+            var durable = written
+                .Where(n => !string.IsNullOrEmpty(n.Path))
+                .GroupBy(n => n.Path, StringComparer.OrdinalIgnoreCase)
+                .ToDictionary(g => g.Key, g => g.First(), StringComparer.OrdinalIgnoreCase);
+            var refused = requested
+                .Where(n => durable.TryGetValue(n.Path, out var durableNode)
+                            && durableNode.Version > n.Version)
+                .ToArray();
+            if (refused.Length == 0)
+                return Observable.Return(
+                    (IReadOnlyDictionary<string, bool>)ImmutableDictionary<string, bool>.Empty);
+
+            logger?.LogWarning(
+                "[PackageInstaller] {Package}: storage REFUSED the bulk write of {Count} node(s) — a "
+                + "durable row newer than the version the package file carries already existed, so the "
+                + "authored content was dropped. Re-deciding them as existing nodes through the "
+                + "validating request path: {Paths}",
+                manifest.Id, refused.Length, string.Join(", ", refused.Select(n => n.Path)));
+
+            return refused
+                .Select(n => DecideAndWrite(hub, durable[n.Path], n, options)
+                    .Catch<bool, Exception>(_ => Upsert(hub, n).Select(_ => true))
+                    .Select(wrote => (n.Path, wrote)))
+                .ToObservable().Concat().ToList()
+                .Select(outcomes => (IReadOnlyDictionary<string, bool>)outcomes
+                    .ToImmutableDictionary(o => o.Path, o => o.wrote, StringComparer.Ordinal));
+        }
+
         // The same per-node type-existence rule the create path applies (MeshExtensions, step 3:
         // static registry → persistence), evaluated once per DISTINCT type instead of once per
         // node — with the types THIS package installs satisfied by construction (the bulk write

test/MeshWeaver.PluginCatalog.Test/BulkSaveInstallTest.cs

@@ -159,6 +159,94 @@ public async Task WhenTheBulkReadFails_EveryNodeFallsBackToTheRequestPath()
         (await Read("BulkPack/Deep")).NodeType.Should().Be("BulkPack/Thing");
     }
 
+    /// <summary>
+    /// A bulk write the STORE REFUSED must never be reported as written — it must be re-decided
+    /// as what it turned out to be, an EXISTING node, so the package's content actually lands
+    /// (MeshWeaver#2361).
+    ///
+    /// <para><b>The defect.</b> The bulk path is chosen for nodes the install's ONE bulk read
+    /// reported absent, and it writes them straight to storage at the version the repo file
+    /// carries — 0. If the path exists by the time the batch lands (any writer that minted a
+    /// version in the window between that read and the write), version 0 is a BACKWARD write:
+    /// the store's version-conditional upsert refuses it and — per
+    /// <c>IStorageAdapter.Write</c>'s contract — emits the DURABLE node instead of the one it
+    /// was handed. The count still matches, so the installer reported success for a write that
+    /// never happened, the partition kept the foreign content, and the only thing that ever
+    /// noticed was the plugin gate's idempotence pin on the NEXT install — where the node now
+    /// exists, takes the request path and finally lands: <c>re-install of the unchanged snapshot
+    /// wrote 2 node(s): Skill/presentation, Skill/slide</c>, intermittently, on a required check.
+    /// </para>
+    ///
+    /// <para>Both assertions below are the invariant, from the two sides that matter: the
+    /// package's content is what is durable after ONE install, and a re-install of the unchanged
+    /// snapshot writes nothing. Before the fix both fail — the first because the foreign row
+    /// survived, the second because the re-install is what repairs it.</para>
+    /// </summary>
+    [Fact(Timeout = 120_000)]
+    public async Task ABulkWriteTheStoreRefused_IsReapplied_NotReportedAsWritten()
+    {
+        Func<string, string, string?, string, IObservable<RepoSnapshot>> fetch =
+            (_, _, _, _) => Observable.Return(new RepoSnapshot("commit-refused", Repo));
+        var source = new NodeRepoPackageSource(fetch, "https://github.com/acme/bulk");
+        var manifest = new PackageManifest
+        {
+            Id = "BulkPack",
+            Name = "Bulk Pack",
+            Kind = PackageKind.NodeRepo,
+            TargetPartition = "BulkPack",
+            SourceFolder = "BulkPack",
+            Version = "commit-refused",
+        };
+        var files = await source.FetchPackageFiles(manifest, "HEAD").FirstAsync().ToTask();
+
+        // Another writer got there first — at a version the install's version-0 bulk write
+        // cannot beat. This is the ONLY thing the test arranges; everything else is the
+        // installer's own behaviour.
+        _recorder.RaceWriteBeforeBatch = new MeshNode("Lesson2", "BulkPack")
+        {
+            Name = "Foreign Lesson",
+            NodeType = "Markdown",
+            State = MeshNodeState.Active,
+            Version = 7,
+            Content = new MeshWeaver.Markdown.MarkdownContent { Content = "# Foreign" },
+        };
+
+        var result = await PackageInstaller.Install(Mesh, manifest, files, "commit-refused")
+            .FirstAsync().Timeout(TimeSpan.FromSeconds(90)).ToTask();
+        _recorder.WriteManyBatches.SelectMany(b => b).Should().Contain("BulkPack/Lesson2",
+            "this test is only meaningful while that node travels the bulk path");
+        result.Total.Should().Be(7);
+
+        // 1. The package's content is DURABLE after one install — the whole point of installing.
+        var stored = await Observable.Interval(TimeSpan.FromMilliseconds(50)).StartWith(0L)
+            .SelectMany(_ => ((IStorageAdapter)_recorder)
+                .Read("BulkPack/Lesson2", Mesh.JsonSerializerOptions))
+            .Where(n => n is not null && !string.Equals(n.Name, "Foreign Lesson", StringComparison.Ordinal))
+            .Select(n => n!)
+            .FirstAsync()
+            // Generous, and never a sleep: the re-apply happens INSIDE the Install this test has
+            // already awaited, so all that is being waited out is the owner's 200 ms debounced
+            // persist. On the unfixed installer nothing ever lands, and the Catch turns the
+            // timeout into the assertion below rather than an opaque TimeoutException.
+            .Timeout(TimeSpan.FromSeconds(15))
+            .Catch((Exception _) => ((IStorageAdapter)_recorder)
+                .Read("BulkPack/Lesson2", Mesh.JsonSerializerOptions).Select(n => n!))
+            .ToTask();
+        stored.Name.Should().NotBe("Foreign Lesson",
+            "an install that reports a node as written must have written it — a refused bulk write "
+            + "left the foreign row in place and said nothing");
+        stored.ContentAs<MeshWeaver.Markdown.MarkdownContent>(Mesh.JsonSerializerOptions)!
+            .Content.Should().Contain("Lesson 2");
+
+        // 2. …and the re-install of the unchanged snapshot writes NOTHING. This is the exact
+        // signal the plugin gate reports, and it can only hold if install #1 really landed.
+        var second = await PackageInstaller.Install(Mesh, manifest, files, "commit-refused")
+            .FirstAsync().Timeout(TimeSpan.FromSeconds(90)).ToTask();
+        second.WrittenPaths.Should().BeEmpty(
+            "the re-install must find the package's own content durable — a node the first "
+            + "install only CLAIMED to write is written here instead, which is what the gate sees");
+    }
+
     /// <summary>
     /// A bulk-written node must be ANNOUNCED on the mesh-change feed, exactly like one written
     /// through the per-node request path. The feed is what invalidates the caches that decide
@@ -299,6 +387,18 @@ private sealed class RecordingStorageAdapter : IStorageAdapter
         /// <summary>When set, <see cref="ReadMany"/> errors — simulating a transient bulk-read failure.</summary>
         public bool FailReadMany { get; set; }
 
+        /// <summary>
+        /// Arms the interleaving #2361 is about: the FIRST <see cref="WriteMany"/> batch that
+        /// carries this node's path lands it in the store — at its own (higher) version — just
+        /// BEFORE the batch is written, so the installer's version-0 bulk write meets a newer
+        /// durable row. That is the production window in one deterministic step: the install's
+        /// single bulk read ran hundreds of milliseconds earlier (before the root write, the
+        /// visibility barrier and the access phase), and any writer that mints a version in
+        /// between turns the batch's "this node is NEW" classification into a backward write.
+        /// Fires ONCE — a re-install must meet the normal path.
+        /// </summary>
+        public MeshNode? RaceWriteBeforeBatch { get; set; }
+
         public IObservable<DataChangeNotification> Changes => ((IStorageAdapter)_inner).Changes;
 
         public IObservable<MeshNode?> Read(string path, JsonSerializerOptions options)
@@ -316,7 +416,12 @@ public IObservable<IReadOnlyList<MeshNode>> WriteMany(
             IReadOnlyCollection<MeshNode> nodes, JsonSerializerOptions options)
         {
             _writeManyBatches.Enqueue(nodes.Select(n => n.Path).ToImmutableList());
-            return ((IStorageAdapter)_inner).WriteMany(nodes, options);
+            var race = RaceWriteBeforeBatch;
+            if (race is null || nodes.All(n => n.Path != race.Path))
+                return ((IStorageAdapter)_inner).WriteMany(nodes, options);
+            RaceWriteBeforeBatch = null;                       // once — the re-install is not raced
+            return ((IStorageAdapter)_inner).Write(race, options)
+                .SelectMany(_ => ((IStorageAdapter)_inner).WriteMany(nodes, options));
         }
 
         public IObservable<string> Delete(string path) => ((IStorageAdapter)_inner).Delete(path);