From f828e21aa915429764683c342458b48ba3357282 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= <6334612+rbuergi@users.noreply.github.com> Date: Fri, 23 Oct 2026 15:30:41 +0200 Subject: [PATCH 1/7] fix(release): re-issue an Unavailable release re-cut on the same activation when the store answers again (#7072) A release re-cut refused as Unavailable (the store could not be reached, nothing written) was stamped as unreleasedBuildPath and never re-issued by the activation that wrote the stamp; only another pod's boot or the next compile created a release (76 min to 3 h on memex, Refs #7068). - ReleaseCreateOutcome carries a ReleaseFailureClass, classified at the arm from the typed rejection reason; stamped as NodeTypeDefinition.UnreleasedBuildFailureClass (settle and batch bake). - StoreReachability.StoreAnswersAgain / ReissueWhenStoreAnswers: re-issue the same write when the store is evidenced back (a change-feed gap, or a commit in the partition; after a contradicted commit, gaps only). - LateReleaseAdoption re-issues an Unavailable stamp at the stamped id on the live activation and adopts it. Other classes stay named failures. - Fault injector RefuseWritesAsUnreachable + isolated regression test. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../LateReleaseAdoption.cs | 132 ++++++++- .../NodeTypeBuildState.cs | 62 +++- .../NodeTypeCompilationHelpers.cs | 12 +- .../NodeTypeCompileState.cs | 5 + .../PrebuiltAssemblySeeder.cs | 1 + .../ReleasePostCondition.cs | 18 +- .../Architecture/FaultInjectionHarness.md | 10 + .../Data/Architecture/NodeTypeCompilation.md | 2 +- .../Architecture/ReleaseFailureClasses.md | 82 ++++++ .../NodeTypeDefinition.cs | 16 + .../ReleaseFailureClass.cs | 38 +++ src/MeshWeaver.Hosting/NodeTypeBatchBake.cs | 16 +- .../NodeTypeOperationalContent.cs | 3 + .../StoreReachability.cs | 95 +++++- ...eRecutIsReissuedWhenTheStoreAnswersTest.cs | 274 ++++++++++++++++++ .../MeshWeaver.Compiler.Pipeline.Test.csproj | 3 + .../ReleaseRecutReportsWhyItFailedTest.cs | 39 +++ .../FaultInjectingStorageAdapter.cs | 43 ++- 18 files changed, 834 insertions(+), 17 deletions(-) create mode 100644 src/MeshWeaver.Graph.Contract/ReleaseFailureClass.cs create mode 100644 test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs diff --git a/src/MeshWeaver.Compiler.Pipeline/LateReleaseAdoption.cs b/src/MeshWeaver.Compiler.Pipeline/LateReleaseAdoption.cs index a6bade70c2..2c82806f30 100644 --- a/src/MeshWeaver.Compiler.Pipeline/LateReleaseAdoption.cs +++ b/src/MeshWeaver.Compiler.Pipeline/LateReleaseAdoption.cs @@ -73,6 +73,7 @@ public static class LateReleaseAdoption LatestReleasePath = stamped, UnreleasedBuildPath = null, UnreleasedBuildReason = null, + UnreleasedBuildFailureClass = null, ReleaseNotes = null, }; } @@ -254,6 +255,123 @@ internal static IObservable CompleteInheritedRelease( }); } + /// + /// 🚨 #7072 β€” the stamp THIS activation must re-issue when the store answers again, or null. + /// Pure. + /// + /// The shape, as measured. On memex on 2026-10-22 a managed database restart (#7068) + /// refused every write for about 40 seconds. Three Hosting NodeTypes settled inside it, and both + /// the release create and its same-id re-cut were answered "could not be attempted: the data + /// store was unreachable … retrying the same request (with the same node id) is + /// meaningful". The settle stamped , + /// correctly, and then nothing on that pod ever made the create again: + /// waits for a node nobody is writing, and runs only on + /// the NEXT activation. The pod stayed up and ready all night, so the types advertised builds no + /// release named for 76 minutes (a different pod's boot cut two of them) and 3 hours (the next + /// compile superseded the third, whose stamped id was never created). + /// + /// So a stamp of class is an obligation of + /// the activation that HOLDS it, not only of the next one. The same guards as + /// apply β€” a settled build of this framework whose recorded + /// bytes the stamped id names β€” so a re-issue can only ever mint the release for these bytes, at + /// the id already stamped. Every other class stands: a refusal re-issued would be refused again, + /// and an expiry's create may still land and is adopted by . + /// + /// The definition as the owner currently holds it. + /// The NodeType's path. + /// The live framework build identity. + internal static string? UnavailableObligation( + NodeTypeDefinition? definition, string hubPath, string frameworkVersion) + => string.Equals(definition?.UnreleasedBuildFailureClass, ReleaseFailureClass.Unavailable, + StringComparison.Ordinal) + ? InheritedObligation(definition, hubPath, frameworkVersion) + : null; + + /// + /// #7072 β€” the activation's half of : while the own record + /// carries such a stamp, re-issue the release create at the stamped id once the store is + /// evidenced to answer again ( β€” a + /// change-feed gap, or a commit in this type's partition), and adopt it in one owner write when + /// it lands. Emits the adopted path. + /// + /// 🚨 No timer, no poll, no widened bound: the re-issue fires on a FACT that the store is + /// back, and while it is not, nothing happens. Idempotent by construction: the create is made at + /// the stamped id, so an earlier attempt that did land after all is met as + /// NodeAlreadyExists and adopted (), and the + /// adoption re-checks the stamp in the owner write (). A new stamp supersedes + /// the watch (Switch); a cleared one ends it. + /// + /// The per-NodeType hub. + /// Its workspace. + /// Runs the create and the write as System. + /// Where the outcome is published. + internal static IObservable ReissueWhenTheStoreAnswers( + IMessageHub hub, + IWorkspace workspace, + AccessService? accessService, + ILogger? logger) + { + var hubPath = hub.Address.Path; + var options = hub.JsonSerializerOptions; + var feed = hub.ServiceProvider.GetService(); + var partition = hubPath.Trim('/').Split('/', 2)[0]; + return workspace.GetMeshNodeStream() + .Where(node => node is not null) + .Select(node => (Node: node!, Stamped: UnavailableObligation( + node!.ContentAs(options), hubPath, + NodeTypeCompilationHelpers.FrameworkVersion))) + .DistinctUntilChanged(x => x.Stamped) + .Select(x => + { + if (x.Stamped is not { } stamped + || x.Node.ContentAs(options) is not { } definition) + return Observable.Empty(); + logger?.LogInformation( + "[ReleasePostCondition] {HubPath}: the release at {ReleasePath} was not created " + + "because the store could not be reached ({Reason}). It is re-issued at that id " + + "as soon as the store is evidenced to answer again β€” a change-feed reconnect or a " + + "commit in partition '{Partition}' (#7072).", + hubPath, stamped, definition.UnreleasedBuildReason ?? "(no reason recorded)", partition); + var pending = x.Node with { Content = definition with { RequestedReleaseBy = null } }; + var build = RecordedBuild(definition)!; + return feed + .ReissueWhenStoreAnswers( + partition, + () => accessService.RunAsSystem(() => NodeTypeBuildState.TryCreateReleaseNode( + hub, hubPath, build, pending, definition.LastCompilationActivityPath, + logger, reusePath: stamped)), + outcome => outcome.WasUnavailable, + outcome => logger?.LogWarning( + "[ReleasePostCondition] {HubPath}: re-issuing the release at {ReleasePath} " + + "was answered Unavailable again{Because}. A commit landed in the partition " + + "and still the store never saw this create, so the next re-issue waits for " + + "the change feed to reconnect (#7072).", + hubPath, stamped, outcome.Because)) + .SelectMany(outcome => + { + if (outcome.ReleasePath is not { } landed) + { + // A verdict, not an outage: the stamp stands and is the report; the next + // activation cuts it once more (#6056). + logger?.LogWarning( + "[ReleasePostCondition] {HubPath}: re-issuing the release at " + + "{ReleasePath} after the store answered again did not land{Because}. " + + "The stamp stands (#7072).", + hubPath, stamped, outcome.Because); + return Observable.Empty(); + } + return accessService.RunAsSystem(() => workspace.GetMeshNodeStream() + .Update(current => + Adopt(current.ContentAs(options), landed) is { } adopted + ? current with { Content = adopted } + : current)) + .Take(1) + .Select(_ => landed); + }); + }) + .Switch(); + } + /// /// Installs the watch on a per-NodeType hub, next to the compile and release-request watchers. /// Returns the subscription so the caller can register it for disposal with the hub. @@ -309,6 +427,18 @@ public static IDisposable Install(IMessageHub hub, IWorkspace workspace) hubPath, landed), logger, $"[LateReleaseAdoption] {hubPath}"); - return new CompositeDisposable(completion, landings); + // #7072 β€” a stamp the store never saw is re-issued on THIS activation when the store is back. + var reissue = ActivityControlPlaneExtensions.SubscribeHubWatcher( + hub, + () => ReissueWhenTheStoreAnswers(hub, workspace, accessService, logger), + landed => logger?.LogInformation( + "[ReleasePostCondition] {HubPath}: the store answered again and the release was " + + "re-issued at {ReleasePath} and adopted: latestReleasePath names it and " + + "unreleasedBuildPath is cleared (#7072).", + hubPath, landed), + logger, + $"[ReleaseReissue] {hubPath}"); + + return new CompositeDisposable(completion, landings, reissue); } } diff --git a/src/MeshWeaver.Compiler.Pipeline/NodeTypeBuildState.cs b/src/MeshWeaver.Compiler.Pipeline/NodeTypeBuildState.cs index 941f95ed77..0ffadabc96 100644 --- a/src/MeshWeaver.Compiler.Pipeline/NodeTypeBuildState.cs +++ b/src/MeshWeaver.Compiler.Pipeline/NodeTypeBuildState.cs @@ -104,8 +104,16 @@ public static class NodeTypeBuildState /// stamp it on the node as UnreleasedBuildPath: the one place to look. null only /// for and for a failure that happened before any id existed. /// + /// + /// 🚨 The of (#7072), classified at the + /// arm that SAW the exception β€” never parsed back out of the sentence afterwards. It is what lets + /// the activation holding the stamp re-issue a create the store never saw + /// () once the store answers again, while a refusal + /// stays a named failure. null exactly when is. + /// internal sealed record ReleaseCreateOutcome( - string? ReleasePath, string? Failure, bool Attempted = true, string? AttemptedPath = null) + string? ReleasePath, string? Failure, bool Attempted = true, string? AttemptedPath = null, + string? FailureClass = null) { /// No create was made β€” this compile was not asked to release anything. internal static readonly ReleaseCreateOutcome NotAttempted = new(null, null, Attempted: false); @@ -118,7 +126,8 @@ internal static ReleaseCreateOutcome Landed(string releasePath) => /// The create was attempted and did not land, for β€” /// before any release id existed (the node could not be composed at all). /// Why, in one line an operator can act on. - internal static ReleaseCreateOutcome Failed(string reason) => new(null, reason); + internal static ReleaseCreateOutcome Failed(string reason) => + new(null, reason, FailureClass: ReleaseFailureClass.Faulted); /// The create at was attempted and did not /// confirmably land, for . The path travels with the reason so a @@ -126,7 +135,20 @@ internal static ReleaseCreateOutcome Landed(string releasePath) => /// Why, in one line an operator can act on. /// The release path the attempt was minting. internal static ReleaseCreateOutcome Failed(string reason, string attemptedPath) => - new(null, reason, AttemptedPath: attemptedPath); + new(null, reason, AttemptedPath: attemptedPath, FailureClass: ReleaseFailureClass.Faulted); + + /// The create at did not land, for + /// , of (#7072). + /// Why, in one line an operator can act on. + /// The release path the attempt was minting. + /// A value. + internal static ReleaseCreateOutcome Failed(string reason, string attemptedPath, string failureClass) => + new(null, reason, AttemptedPath: attemptedPath, FailureClass: failureClass); + + /// True when the create was never evaluated by the store β€” re-issuing it at the + /// same id is meaningful once the store answers again (#7072). + internal bool WasUnavailable => + string.Equals(FailureClass, ReleaseFailureClass.Unavailable, StringComparison.Ordinal); /// True when a release exists for these bytes. internal bool Succeeded => ReleasePath is not null; @@ -553,9 +575,41 @@ private static IObservable BoundedFrom( return Observable.Return(ReleaseCreateOutcome.Failed( Describe(ex, releasePath, ex is TimeoutException ? IoPoolQueueReport.Describe(pools, baseline) : null), - releasePath)); + releasePath, + Classify(ex))); }); + /// + /// The of one create failure, read off the exception AT THE ARM + /// that received it (#7072) β€” the typed rejection reason IMeshService.CreateNode stamps on + /// , or the store-unreachable predicate for a driver fault that + /// reached here unwrapped. Pure. + /// + /// 🚨 Classified by the CONDITION, never by the sentence: Unavailable is the reason + /// value that means "not evaluated, nothing written, a retry with the same id is meaningful" + /// (#1446, #3050), and it is the one class the activation holding the stamp re-issues. The bound + /// expiring is , not Unavailable β€” the create may still + /// land, and the landing is adopted when it does. + /// + /// The exception the create failed with. + internal static string Classify(Exception ex) + { + if (ex is TimeoutException) + return ReleaseFailureClass.Expired; + if (ex.Data[NodeCreationFailure.RejectionReasonKey] is NodeCreationRejectionReason reason) + return reason switch + { + NodeCreationRejectionReason.Unavailable => ReleaseFailureClass.Unavailable, + NodeCreationRejectionReason.Unknown => StoreReachability.IsStoreUnreachable(ex) + ? ReleaseFailureClass.Unavailable + : ReleaseFailureClass.Faulted, + _ => ReleaseFailureClass.Refused, + }; + return StoreReachability.IsStoreUnreachable(ex) + ? ReleaseFailureClass.Unavailable + : ReleaseFailureClass.Faulted; + } + /// /// One create failure as one operator-readable line. A is named /// for what it IS β€” the bound expired and the owning hub never answered β€” because "TimeoutException: diff --git a/src/MeshWeaver.Compiler.Pipeline/NodeTypeCompilationHelpers.cs b/src/MeshWeaver.Compiler.Pipeline/NodeTypeCompilationHelpers.cs index 738d1c2b19..da678182c2 100644 --- a/src/MeshWeaver.Compiler.Pipeline/NodeTypeCompilationHelpers.cs +++ b/src/MeshWeaver.Compiler.Pipeline/NodeTypeCompilationHelpers.cs @@ -4586,7 +4586,8 @@ internal static NodeTypeDefinition ApplyCompileSuccess( string? modulesHash = null, string? unreleasedBuildPath = null, string? unreleasedBuildReason = null, - NodeTypeDefinition? compiledInput = null) + NodeTypeDefinition? compiledInput = null, + string? unreleasedBuildFailureClass = null) { var assemblyMvid = ServedBuildIdentity.OfFile(result.AssemblyLocation); var dependencies = result.CompiledDependencies; @@ -4672,6 +4673,11 @@ internal static NodeTypeDefinition ApplyCompileSuccess( UnreleasedBuildReason = releasePath is not null || unreleasedBuildPath is null ? null : unreleasedBuildReason, + // #7072 β€” the CLASS travels with the reason, so the activation holding the stamp can + // tell a create the store never saw (re-issued when it answers again) from a verdict. + UnreleasedBuildFailureClass = releasePath is not null || unreleasedBuildPath is null + ? null + : unreleasedBuildFailureClass, ReleaseNotes = releasePath is not null ? null : def.ReleaseNotes, CompiledSources = result.CompiledSources ?? System.Collections.Immutable.ImmutableDictionary.Empty, @@ -5667,7 +5673,9 @@ IObservable StampCompileState() => // #6831 β€” the record the compile READ, so the bytes carry its // module version and not one that landed while it ran. compiledInput: outcome.PendingNode.ContentAs( - SettleHub().JsonSerializerOptions, logger)); + SettleHub().JsonSerializerOptions, logger), + // #7072 β€” whether the stamp is one the store never saw. + unreleasedBuildFailureClass: settle.UnreleasedBuildFailureClass); lifted = BuildDeliveryHold.EventOf(def, succeeded); liftedDef = succeeded; return curr with { Content = succeeded }; diff --git a/src/MeshWeaver.Compiler.Pipeline/NodeTypeCompileState.cs b/src/MeshWeaver.Compiler.Pipeline/NodeTypeCompileState.cs index c90013e4ce..539b9b255f 100644 --- a/src/MeshWeaver.Compiler.Pipeline/NodeTypeCompileState.cs +++ b/src/MeshWeaver.Compiler.Pipeline/NodeTypeCompileState.cs @@ -102,6 +102,9 @@ public record NodeTypeCompileState /// See . public string? UnreleasedBuildReason { get; init; } + /// See (#7072). + public string? UnreleasedBuildFailureClass { get; init; } + /// See . public string? ReleaseNotes { get; init; } @@ -239,6 +242,7 @@ definition is null LastReleaseRequestHandledAt = definition.LastReleaseRequestHandledAt, UnreleasedBuildPath = definition.UnreleasedBuildPath, UnreleasedBuildReason = definition.UnreleasedBuildReason, + UnreleasedBuildFailureClass = definition.UnreleasedBuildFailureClass, ReleaseNotes = definition.ReleaseNotes, LatestAssemblyCollection = definition.LatestAssemblyCollection, LatestAssemblyPath = definition.LatestAssemblyPath, @@ -279,6 +283,7 @@ CompilationStatus is null && CompilationError is null && CompilationDiagnostics && RequestedReleaseAt is null && !RequestedReleaseForce && RequestedReleaseBy is null && LastReleaseRequestHandledAt is null && ReleaseNotes is null && UnreleasedBuildPath is null && UnreleasedBuildReason is null + && UnreleasedBuildFailureClass is null && LatestAssemblyCollection is null && LatestAssemblyPath is null && LatestAssemblyMvid is null && CompiledSources is null && CurrentSourceVersions is null diff --git a/src/MeshWeaver.Compiler.Pipeline/PrebuiltAssemblySeeder.cs b/src/MeshWeaver.Compiler.Pipeline/PrebuiltAssemblySeeder.cs index fc10d25e93..edb1daf0a3 100644 --- a/src/MeshWeaver.Compiler.Pipeline/PrebuiltAssemblySeeder.cs +++ b/src/MeshWeaver.Compiler.Pipeline/PrebuiltAssemblySeeder.cs @@ -1389,6 +1389,7 @@ MeshNode Apply(MeshNode? current) // of the SAME coordinates still owes its pending release, so keep it. UnreleasedBuildPath = retirePendingRelease ? null : def.UnreleasedBuildPath, UnreleasedBuildReason = retirePendingRelease ? null : def.UnreleasedBuildReason, + UnreleasedBuildFailureClass = retirePendingRelease ? null : def.UnreleasedBuildFailureClass, // The adopted bytes' own identity (#2471), read from the image // in hand β€” no file, no load. An adopted build is exactly the // case where a path says least: several pods adopt the same diff --git a/src/MeshWeaver.Compiler.Pipeline/ReleasePostCondition.cs b/src/MeshWeaver.Compiler.Pipeline/ReleasePostCondition.cs index 82e6cceab2..c275dd20a0 100644 --- a/src/MeshWeaver.Compiler.Pipeline/ReleasePostCondition.cs +++ b/src/MeshWeaver.Compiler.Pipeline/ReleasePostCondition.cs @@ -167,11 +167,17 @@ private static bool SameSnapshot( /// release exists, and whenever the settle was not a violation at all. /// /// Why, in the re-cut's own words; null with the path. + /// + /// The of that reason (#7072) β€” stamped beside it, so the + /// activation holding the stamp re-issues a create the store never saw once the store answers + /// again, and leaves every other class standing. null with the path. + /// internal sealed record Settle( string? ReleasePath, LogMessage? Diagnosis, string? UnreleasedBuildPath = null, - string? UnreleasedBuildReason = null); + string? UnreleasedBuildReason = null, + string? UnreleasedBuildFailureClass = null); /// /// The settle-path remedy. Emits the release path the terminal stamp should use: the one this @@ -323,7 +329,8 @@ internal static IObservable Restore( return new Settle( null, AbandonedEntry(violation, firstFailure, recut), UnreleasedBuildPath: unreleased, - UnreleasedBuildReason: AbandonedReason(recut)); + UnreleasedBuildReason: AbandonedReason(recut), + UnreleasedBuildFailureClass: recut.FailureClass); } logger?.LogError( @@ -338,7 +345,12 @@ internal static IObservable Restore( UnreleasedBuildPath: unreleased, UnreleasedBuildReason: recut.Failure is { Length: > 0 } reason ? reason - : "the re-cut reported no reason, which is itself a defect in this pipeline"); + : "the re-cut reported no reason, which is itself a defect in this pipeline", + // 🚨 #7072 β€” an Unavailable re-cut is NOT the end of the story on this pod: the + // store never saw the create, and the activation holding this stamp re-issues it + // at the same id once the store is evidenced to answer again + // (LateReleaseAdoption.ReissueWhenTheStoreAnswers). + UnreleasedBuildFailureClass: recut.FailureClass ?? ReleaseFailureClass.Faulted); }); } diff --git a/src/MeshWeaver.Documentation/Data/Architecture/FaultInjectionHarness.md b/src/MeshWeaver.Documentation/Data/Architecture/FaultInjectionHarness.md index 97b27ea2ed..c10fee5882 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/FaultInjectionHarness.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/FaultInjectionHarness.md @@ -113,6 +113,15 @@ if the path did not exist yet; writes still land. The shape of a resolver that h with a create another process acknowledged. Resolution stops at the parent, so the router answers `No node found at '…'` exactly as production logs it. +### Unreachable store for one path + +`RefuseWritesAsUnreachable(path)`: every write of the path (`Write`, `WriteIfVersion`, a `WriteMany` +naming it) faults with a driver connection failure, a `DbException` with SQLSTATE `08006`, and +writes nothing. The create handler's store-unreachable arm answers `Unavailable`, which is the +production answer during a managed database restart (#7068). Releasing the switch is the store +coming back. Pair it with `DeliverChangeFeedGap(...)`, which is what the LISTEN reconnect declares +after such a restart. + ### Change-feed hold `HoldChangeFeed()`: the silo's change notifications β€” its own commits and relayed ones β€” queue in @@ -152,6 +161,7 @@ sweep re-fed exactly what the sender lost. | 4 | Steward's first write after its create (Plugins#2530, #6045, #6046) | `ARoutedWriteRightAfterItsCreateTest` (6 cases) | `HidePath`, `Relay.Hold`, a real pre-create probe | `MeshNodeStreamCache.ResetFailureState` made a no-op β†’ the probe case red: `No node found at '…/Item'`; the read-window write fast-fail restored in `UpdateRaw` β†’ the cross-replica case red with the same line; the cached-remainder route restored in `PathResolutionService` β†’ the existing-parent case red: `… Closest ancestor is '…' (remainder='Item')` | | 6 | Fleet watch freeze (#5011) | `AHeldReadSurvivesItsSourceSilo{BeingKilled,Draining}Test`, `AHeldReadOnAThirdSilo…`, `AHeldReadFollowsItsOwnersHandOffTest`, `AHeldReadFollowsItsOwnersHandOffWithoutTheChangeFeedTest` | `Kill`, `Drain`, `Linger`, `HoldChangeFeed`, `HandOffTarget` | held-stream heartbeat pushed beyond the budget β†’ all four kill/drain cases red (the owner never re-activates); the owner's answer to a heartbeat for a stream it does not serve removed (`MeshExtensions.HandleHeartBeat`) β†’ the withheld-feed hand-off case red: the held read emits nothing | | 7 | Resumer hold without a heartbeat (Plugins#2403, #6048), core half | `AHeldReadUnderAShortIdleWindowSurvivesItsOwnerSiloBeingKilledTest`, `AHandleHeldReadUnderAShortIdleWindow…`, `AReadThenHoldAcrossTheIdleRelease…` | `Kill`, idle sweep 500 ms / 100 ms (`ShortIdleSiloConfigurator`) | `Entry.IsIdleCandidate`/`TryMarkIdleEvicted` ignoring the subscriber count β†’ red after 36 s: the owner never re-activates, "emitted nothing at all" | +| 9 | A release re-cut refused during a database restart, never re-issued on the live pod (#7072) | `AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest` (MeshWeaver.Compiler.Pipeline.Test) | `RefuseWritesAsUnreachable`, `DeliverChangeFeedGap` | the re-issue watch removed from `LateReleaseAdoption.Install` β†’ red at the final wait: the pointer never reaches the stamped id; a `Refused` stamp under the same signals is never re-issued | Cases 2, 3, 5, 7 and 8 exercise code that lives in MeshWeaver.Plugins, which reaches this harness by `ProjectReference` through its platform checkout (`Requires-platform: MeshWeaver#5879`): diff --git a/src/MeshWeaver.Documentation/Data/Architecture/NodeTypeCompilation.md b/src/MeshWeaver.Documentation/Data/Architecture/NodeTypeCompilation.md index 64221a5a33..041e73c5a7 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/NodeTypeCompilation.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/NodeTypeCompilation.md @@ -3504,7 +3504,7 @@ protocol lives in the `/code` skill. | Cancel a running compile | `hub.CancelActivity(activityPath)` | | List releases | The `Release/*` children of the NodeType | | Find the current release | `NodeTypeDefinition.LatestReleasePath` | -| Tell whether the LATEST build has a release at all | `NodeTypeDefinition.UnreleasedBuildPath` β€” null when it does; otherwise the id the create was minting, with `UnreleasedBuildReason` beside it (#5057, [Release Failure Classes](../ReleaseFailureClasses)) | +| Tell whether the LATEST build has a release at all | `NodeTypeDefinition.UnreleasedBuildPath` β€” null when it does; otherwise the id the create was minting, with `UnreleasedBuildReason` and its class `UnreleasedBuildFailureClass` beside it; an `Unavailable` one is re-issued by the holding activation once the store answers again (#5057, #7072, [Release Failure Classes](../ReleaseFailureClasses)) | | Pin instances to a fixed release | Set `NodeTypeDefinition.RequestedReleasePath` | | Understand why it recompiled | `HasUsableBuild` failed rule 2 (assembly gone) or rule 3 (framework changed) | | Understand a "Compile leg '…' did not complete within Ns" error | That stage stopped answering β€” see [Every stage is bounded](#every-stage-is-bounded--a-compile-can-never-park-at-compiling) | diff --git a/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md b/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md index 936a519e0b..f9fb297e82 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md @@ -539,6 +539,88 @@ SIGTERM on that replica against those two creates. The first attempts' failures partition-root creates. #6088 stops a root-read fault from being misread as absence. Why those reads faulted is still not recorded. +## A re-cut the store never saw is re-issued when the store answers again (#7072) + +### What was still live after the next-activation obligation + +The next activation honours a stamp, but nothing honoured it on the activation that WROTE it. On memex +on 2026-10-22 a managed database restart (#7068) refused writes for about 40 seconds. Between +21:22:55Z and 21:23:04Z three Hosting NodeTypes settled on pod `85cf49759f-77plc`. For `PrWatch`, +`TriageItem` and `TriageStatus`, the release create and its same-id re-cut were both answered +`[CreateNode] STORE UNREACHABLE` β†’ *"could not be attempted: the data store was unreachable … retrying +the same request (with the same node id) is meaningful"*. Each settle stamped `unreleasedBuildPath`. +The store answered again at about 21:23:34Z. The pod stayed up and ready all night, and in the next +76 minutes no line at all mentions these three types: + +- late adoption re-subscribed after its query faulted (`[LateReleaseAdoption] … faulted … + re-establish`), but it can only adopt a node somebody writes; +- the next-activation obligation runs on the FIRST emission of a NEW activation; +- so the release came from unrelated events. A replacement pod's boot cut two of them at 22:39:16Z. + `TriageItem`'s next compile, at 00:18Z, minted a NEW release, and its stamped id was never created. + +The log line said the retry was meaningful, and no component on the live pod made it. + +### The change: the class is stamped, and an `Unavailable` stamp is re-issued on a FACT + +1. **The failure carries a class.** `ReleaseCreateOutcome.FailureClass` is a `ReleaseFailureClass` + (`Unavailable`, `Expired`, `Refused`, `Faulted`: an open set of string constants). It is + classified in `NodeTypeBuildState.Classify` at the arm that received the exception, from the typed + `NodeCreationRejectionReason` that `IMeshService.CreateNode` puts on `Exception.Data`. The sentence + is never parsed. The settle stamps it as `NodeTypeDefinition.UnreleasedBuildFailureClass` beside + the path and the reason, the batch bake does the same, and adoption clears it with them. +2. **The activation that holds an `Unavailable` stamp re-issues it.** `LateReleaseAdoption.Install` + runs `ReissueWhenTheStoreAnswers`. `UnavailableObligation` is its pure gate: the + `InheritedObligation` guards (settled, this framework, the stamped id names the recorded bytes) + plus class `Unavailable`. The re-issue is the same `TryCreateReleaseNode` at the SAME id, as + System, and a landing is adopted in one owner write. +3. **It waits for evidence, never for time.** `StoreReachability.ReissueWhenStoreAnswers` in + `MeshWeaver.Mesh.Contract` waits for `StoreAnswersAgain`. That signal fires on a declared + `ChangeFeedGap`, which is the LISTEN connection resuming after a database restart, or on a commit + announced in the type's partition, which the store must have accepted. While the store is away, + neither fires and nothing is re-issued. If a re-issue is answered `Unavailable` again after a + partition commit, that commit has been shown not to be evidence for this write. Later re-issues + then wait for a feed reconnect only. A busy partition therefore cannot turn this into a retry + loop, and no count or clock is involved. + +Every other class stays standing. A `Refused` create would be refused again. An `Expired` one may +still land, and late adoption takes it when it does. A stamp written before the class existed has +no class and is left to the next activation, as before. + +**Why this signal and not a store health probe or a timer.** A timer that retries "in case the store +is back" is the band-aid `StoreReachability` already refuses, and on a store that is the bottleneck +it adds load at the worst moment. A health probe would be a second, weaker opinion of what the +LISTEN connection and the commit feed already report as facts. Both signals are mesh-scoped, already +consumed by the path-resolution and stream caches for the same reason, and silent exactly as long as +a re-issue would be pointless. + +**It generalises.** `ReissueWhenStoreAnswers(feed, partition, attempt, wasUnavailable)` is the one +shape for any write answered `Unavailable`: the same request with the same id, re-issued when the +store is evidenced back. The release re-cut is its first caller. Other sites of the same class can +adopt it in their own changes, for example the control-instance finding +`rbuergi/Feedback/unavailable-write-retried-76-min-later`. + +### The control + +`AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest` (MeshWeaver.Compiler.Pipeline.Test) +runs on a real monolith mesh with the shared fault injector. `RefuseWritesAsUnreachable` answers +every write of the ONE release path with a driver connection failure (SQLSTATE `08006`). The owner +is activated before any stamp exists, so the next-activation obligation cannot be what repairs it. +The re-cut runs through `TryCreateReleaseNode` and must come back `Unavailable`. The stamp is written +through `ApplyCompileSuccess`. While the store still refuses, the pointer must not move. Releasing +the switch and declaring a feed gap must then leave `latestReleasePath` on the ORIGINAL stamped id, +with the stamp, its reason and its class cleared and a release node there for the recorded build. +`ARefusedRecut_IsNeverReissued_HoweverOftenTheStoreAnswers` is the negative control: a `Refused` +stamp under the same signals is never re-issued. With the re-issue watch removed from `Install`, +the main case is red at its final wait. `ReleaseRecutReportsWhyItFailedTest` pins the +classification through the real `Bounded` chain. + +### Not established + +The memex pod's own `StoreAnswersAgain` evidence for that night was not read. Whether its LISTEN +connection dropped and declared a gap during the 40-second refusal is inferred from how a managed +restart behaves, not from a log line. A commit in `Hosting` after 21:23:34Z would have triggered the +re-issue either way. + ## A re-cut cut short by the process's own shutdown is a Warning (#6056, second half) ### What the logs and the version history show diff --git a/src/MeshWeaver.Graph.Contract/NodeTypeDefinition.cs b/src/MeshWeaver.Graph.Contract/NodeTypeDefinition.cs index 75f896f8f8..bd29289a1c 100644 --- a/src/MeshWeaver.Graph.Contract/NodeTypeDefinition.cs +++ b/src/MeshWeaver.Graph.Contract/NodeTypeDefinition.cs @@ -549,6 +549,22 @@ public record NodeTypeDefinition /// public string? UnreleasedBuildReason { get; init; } + /// + /// 🚨 The CLASS of β€” a + /// value β€” so the mesh can act on the stamp without parsing a sentence (#7072). null + /// whenever is, and for a stamp written before the class + /// existed. + /// + /// It decides one thing: a stamp of class + /// (the store could not be reached, nothing was written) is RE-ISSUED at the stamped id by the + /// activation that holds it, once the store is evidenced to answer again β€” a declared change-feed + /// gap or a commit in the type's partition. Before this, such a stamp stood until another pod + /// booted or the next compile ran: measured on memex on 2026-10-22, 76 minutes to 3 hours after + /// the store had recovered. Any other class is never re-issued on that activation. + /// Mesh-owned, never authored. + /// + public string? UnreleasedBuildFailureClass { get; init; } + /// /// Content-collection name where the latest compiled assembly for this NodeType /// lives (e.g. "nodetype-cache"). Pair with diff --git a/src/MeshWeaver.Graph.Contract/ReleaseFailureClass.cs b/src/MeshWeaver.Graph.Contract/ReleaseFailureClass.cs new file mode 100644 index 0000000000..9ae5b0733a --- /dev/null +++ b/src/MeshWeaver.Graph.Contract/ReleaseFailureClass.cs @@ -0,0 +1,38 @@ +namespace MeshWeaver.Graph.Configuration; + +/// +/// WHY a release create did not land, as a CLASS a reader can act on β€” the value of +/// (#7072). +/// +/// 🚨 An OPEN vocabulary of string constants, never an enum (policy +/// open-vocabulary-string-constants): the field is persisted on every NodeType record, so an +/// enum deserialising a value it does not know would fall to its zero member and claim a class it +/// never measured. A value not listed here is UNKNOWN and is acted on by nothing. +/// +/// The class decides what the mesh does next, which is why it is stamped rather than parsed +/// out of : only is +/// re-issued on the activation that holds the stamp, and only once the store is evidenced to answer +/// again β€” a refusal re-issued would be a retry loop on a verdict. See +/// Doc/Architecture/ReleaseFailureClasses. +/// +public static class ReleaseFailureClass +{ + /// + /// The create was NOT ATTEMPTED: the store could not be reached, or the create was answered + /// Unavailable for another not-evaluated reason. Nothing was written, and re-issuing the + /// same create at the same id is meaningful once the store answers again. + /// + public const string Unavailable = "Unavailable"; + + /// + /// The settle stopped WAITING before an answer came. The create may still land, and the landing + /// is adopted when it does (LateReleaseAdoption.Landings). + /// + public const string Expired = "Expired"; + + /// The create was judged and REFUSED. Re-issuing it would be refused again. + public const string Refused = "Refused"; + + /// The create failed for a reason none of the classes above names. + public const string Faulted = "Faulted"; +} diff --git a/src/MeshWeaver.Hosting/NodeTypeBatchBake.cs b/src/MeshWeaver.Hosting/NodeTypeBatchBake.cs index 35628e4add..cdcb9f4c08 100644 --- a/src/MeshWeaver.Hosting/NodeTypeBatchBake.cs +++ b/src/MeshWeaver.Hosting/NodeTypeBatchBake.cs @@ -1343,6 +1343,11 @@ private static IObservable StampAndReport( : null, unreleasedBuildReason: outcome is { Attempted: true, Succeeded: false } ? outcome.Failure + : null, + // #7072 β€” the class, so the activation that holds this stamp can tell a create the + // store never saw from a verdict. + unreleasedBuildFailureClass: outcome is { Attempted: true, Succeeded: false } + ? outcome.FailureClass : null)) .Select(_ => { @@ -1468,12 +1473,13 @@ internal static IObservable WriteStamp( DateTimeOffset startedAt, ILogger? logger, string? unreleasedBuildPath = null, - string? unreleasedBuildReason = null) + string? unreleasedBuildReason = null, + string? unreleasedBuildFailureClass = null) { var gate = mesh.ServiceProvider.GetService(); IObservable Stamp() => BuildStamp(mesh, typeNode, ok, result, error, releasePath, startedAt, logger, - unreleasedBuildPath, unreleasedBuildReason); + unreleasedBuildPath, unreleasedBuildReason, unreleasedBuildFailureClass); return gate is null ? Observable.Defer(Stamp) : gate.Publish($"NodeType compile stamp for {typeNode.Path}", Stamp); @@ -1493,7 +1499,8 @@ private static IObservable BuildStamp( DateTimeOffset startedAt, ILogger? logger, string? unreleasedBuildPath, - string? unreleasedBuildReason) + string? unreleasedBuildReason, + string? unreleasedBuildFailureClass) { var storage = mesh.ServiceProvider.GetService(); if (storage is null) @@ -1556,7 +1563,8 @@ IObservable BuildStampAt(ImmutableList? refusals) => storag unreleasedBuildPath, unreleasedBuildReason, // #6831 β€” the node the compiler was handed, whose module version // the bytes carry. - compiledInput: typeNode.ContentAs(options, logger)) + compiledInput: typeNode.ContentAs(options, logger), + unreleasedBuildFailureClass: unreleasedBuildFailureClass) : NodeTypeCompilationHelpers.ApplyCompileFailure( def, result, error, activityPath: null, mesh.ServiceProvider.GetService()?.Hash, diff --git a/src/MeshWeaver.Mesh.Contract/NodeTypeOperationalContent.cs b/src/MeshWeaver.Mesh.Contract/NodeTypeOperationalContent.cs index 571b7bb9d8..29c7b15fd3 100644 --- a/src/MeshWeaver.Mesh.Contract/NodeTypeOperationalContent.cs +++ b/src/MeshWeaver.Mesh.Contract/NodeTypeOperationalContent.cs @@ -91,6 +91,9 @@ public static bool IsNodeTypeNode(MeshNode? node) => // null would erase the one readable trace of a release create that expired its bound. "unreleasedBuildPath", "unreleasedBuildReason", + // #7072 β€” the CLASS of that reason; it decides whether the activation holding the stamp + // re-issues the create when the store answers again. Operational for the same reason. + "unreleasedBuildFailureClass", "releaseNotes", "latestAssemblyCollection", "latestAssemblyPath", diff --git a/src/MeshWeaver.Mesh.Contract/StoreReachability.cs b/src/MeshWeaver.Mesh.Contract/StoreReachability.cs index 6497428113..fbe91fe762 100644 --- a/src/MeshWeaver.Mesh.Contract/StoreReachability.cs +++ b/src/MeshWeaver.Mesh.Contract/StoreReachability.cs @@ -1,4 +1,7 @@ -ο»Ώusing MeshWeaver.Data; +ο»Ώusing System.Reactive; +using System.Reactive.Linq; +using MeshWeaver.Data; +using MeshWeaver.Mesh.Services; namespace MeshWeaver.Mesh; @@ -94,6 +97,96 @@ public static string DescribeMayHavePartiallyLanded(string operation) + "not a refusal β€” but read the current state before retrying, because re-sending the " + "whole request could duplicate whatever landed."; + // ── What a caller DOES with "retrying is meaningful" (#7072) ───────────────────────────────── + // + // The sentence above tells a caller that a retry with the same id is meaningful, and until #7072 + // nothing in the platform said WHEN. A NodeType release re-cut refused this way during a 40-second + // database restart (#7068) was never re-issued by the pod that stamped it, although the pod stayed + // up and ready all night: the release came 76 minutes to 3 hours later, from an unrelated event. + // The answer is not a timer that retries in case the store is back (the band-aid this class + // already refuses above), but a FACT that it is. + + /// + /// The EVIDENCE that the store answers again, as this process can observe it (#7072): each + /// declared β€” the backend's cross-process feed lost its connection + /// and RESUMED, which a database restart always produces β€” and each commit announced in + /// , since a commit is announced only after the store accepted it. + /// + /// 🚨 Facts, never a clock. Nothing here fires because time passed; while the store is + /// unreachable no commit lands and no feed resumes, so the signal stays silent exactly as long as + /// re-issuing a write would be pointless. Never emits on a host without an invalidation feed β€” + /// such a host has one process and no cross-process store to lose. + /// + /// The mesh's invalidation feed, or null. + /// The partition whose commits count (its first path segment), or + /// null to count every commit. + /// An infinite signal; one element per piece of evidence. + public static IObservable StoreAnswersAgain(this IMeshInvalidationFeed? feed, string? partition) + { + if (feed is null) + return Observable.Never(); + // The handler is forwarded as-is and the filter runs as an operator, so a fault in it reaches + // the subscriber's OnError instead of being thrown out of the feed's delivery callback. + var commits = Observable.Create(observer => feed.Subscribe(observer.OnNext)) + .Where(change => partition is null || InPartition(change.Path, partition)) + .Select(_ => Unit.Default); + return commits.Merge(feed.Gaps.Select(_ => Unit.Default)); + } + + /// + /// Re-issues a write that was answered Unavailable β€” "not attempted, nothing written, + /// retrying the same request is meaningful" β€” once the store is evidenced to answer again (#7072). + /// The ONE shape for every such site: the caller supplies the SAME request (same node id), how + /// to tell an Unavailable outcome, and the partition the write lands in. + /// + /// How it waits. The first re-issue waits for β€” + /// a feed that resumed, or a commit in the partition. If THAT re-issue is answered Unavailable + /// again, a commit in the partition has just been shown not to be evidence for this write (the + /// store took someone else's commit and still never saw ours), so every later re-issue waits for + /// the feed to RESUME instead β€” a reconnect, which is rare. That is what keeps this from turning a + /// busy partition into a retry loop, without any count or clock. + /// + /// What it never does. It never re-issues a write that was JUDGED: an outcome + /// rejects is emitted and the sequence completes, so a refusal + /// stays a named failure. And it never issues the FIRST attempt β€” the caller made it, and is here + /// because it came back Unavailable. + /// + /// The write's outcome. + /// The mesh's invalidation feed, or null (then nothing is re-issued). + /// The partition the write lands in. + /// The SAME write, cold β€” subscribed once per re-issue. + /// True when an outcome means "not attempted, retry meaningful". + /// Told about each re-issue that was answered Unavailable again. + /// The first outcome that is not Unavailable; never completes while the store stays away. + public static IObservable ReissueWhenStoreAnswers( + this IMeshInvalidationFeed? feed, + string partition, + Func> attempt, + Func wasUnavailable, + Action? onStillUnavailable = null) + { + if (feed is null) + return Observable.Never(); + var reconnects = feed.Gaps.Select(_ => Unit.Default); + IObservable After(IObservable evidence) => + evidence + .Take(1) + .SelectMany(_ => Observable.Defer(attempt).Take(1)) + .SelectMany(outcome => + { + if (!wasUnavailable(outcome)) + return Observable.Return(outcome); + onStillUnavailable?.Invoke(outcome); + return Observable.Defer(() => After(reconnects)); + }); + return Observable.Defer(() => After(feed.StoreAnswersAgain(partition))); + } + + private static bool InPartition(string? path, string partition) + => !string.IsNullOrEmpty(path) + && string.Equals(path.Trim('/').Split('/', 2)[0], partition.Trim('/'), + StringComparison.OrdinalIgnoreCase); + // ── The same verdicts, keyed (#4507) ─────────────────────────────────────────────────────── // // 🚨 The ENGLISH half is composed by the two methods above, never retyped. That is what keeps diff --git a/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs b/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs new file mode 100644 index 0000000000..e00b83fd50 --- /dev/null +++ b/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs @@ -0,0 +1,274 @@ +using System; +using System.Collections.Immutable; +using System.IO; +using System.Reactive.Linq; +using System.Threading.Tasks; +using MeshWeaver.Data; +using MeshWeaver.Fixture; +using MeshWeaver.Graph; +using MeshWeaver.Graph.Configuration; +using MeshWeaver.Hosting.Monolith.TestBase; +using MeshWeaver.Mesh; +using MeshWeaver.Mesh.Services; +using MeshWeaver.Messaging; +using MeshWeaver.Testing.FaultInjection; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Xunit; + +namespace MeshWeaver.Compiler.Pipeline.Test; + +/// +/// 🚨 Issue #7072: a release re-cut the store never saw is re-issued on the SAME activation once +/// the store answers again. +/// +/// The incident. On memex on 2026-10-22 a managed database restart (#7068) refused writes +/// for about 40 seconds. Three Hosting NodeTypes settled inside that window; their release create and +/// its same-id re-cut were both answered "could not be attempted: the data store was unreachable … +/// retrying the same request (with the same node id) is meaningful", and the settle stamped +/// unreleasedBuildPath. The pod stayed up and ready, and nothing on it ever made that create +/// again: the types advertised builds no release named for 76 minutes and 3 hours, until another pod +/// booted or the next compile ran. +/// +/// The isolated fault. +/// on the ONE release path β€” a driver connection failure (SQLSTATE 08006), so the create handler's own +/// store-unreachable arm answers Unavailable, the production answer. Nothing else is perturbed. +/// The owner is activated BEFORE the stamp exists, so the next-activation obligation +/// () cannot be what repairs it: only the +/// activation that holds the stamp can. Releasing the switch is the store coming back, and the +/// declared change-feed gap is what a PostgreSQL LISTEN reconnect announces after a restart. +/// +/// Negative controls. (1) While the store still refuses, the stamp stands β€” the re-issue +/// is not a loop that lands by persistence. (2) A stamp of class +/// is never re-issued, however often the store answers. (3) Recorded on the PR: with the re-issue +/// watch not installed, +/// fails at its final Match, because nothing else on this activation ever creates the node. +/// +public class AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest(ITestOutputHelper output) + : MonolithMeshTestBase(output) +{ + private const string PreviousRelease = "Release/20261020115342-L-vbFJob"; + + private FaultInjectingStorageAdapter Storage => Mesh.ServiceProvider.GetRequiredService(); + + private IMeshService MeshService => Mesh.ServiceProvider.GetRequiredService(); + + /// The injector goes in BEFORE the base adds in-memory persistence. + protected override MeshBuilder ConfigureMesh(MeshBuilder builder) + => base.ConfigureMesh(builder.AddFaultInjectingStorage()); + + /// + /// A settled, usable build of THIS framework whose bytes are in the assembly store, with NO stamp, + /// its owner activated. Returns the build and the release path the settle would mint for it. + /// + private async Task<(NodeCompilationResult Build, string Stamped, MeshNode Seeded)> SeedActivatedAsync(string typePath) + { + var ct = TestContext.Current.CancellationToken; + var bytes = File.ReadAllBytes(typeof(AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest).Assembly.Location); + var location = await Mesh.ServiceProvider.GetRequiredService() + .PutWithLocation(typePath, 736, bytes, null) + .Should().Within(TestTimeouts.WriteConvergence).Emit(cancellationToken: ct); + var definition = new NodeTypeDefinition + { + Configuration = "config => config", + CompilationStatus = CompilationStatus.Ok, + LastCompiledVersion = 736, + LatestReleasePath = $"{typePath}/{PreviousRelease}", + LatestAssemblyCollection = location.Collection, + LatestAssemblyPath = location.ContentPath, + CompiledFrameworkVersion = NodeTypeCompilationHelpers.FrameworkVersion, + CompiledSources = ImmutableDictionary.Empty, + }; + await MeshService.CreateNode(MeshNode.FromPath(typePath) with + { + Name = typePath, + NodeType = MeshNode.NodeTypePath, + State = MeshNodeState.Active, + Content = definition, + }) + .Should().Within(TestTimeouts.Convergence) + .Emit("the NodeType must exist", cancellationToken: ct); + + // Reading the node through the stream activates its owner, and with it the watchers that + // MeshDataSource installs on every NodeType hub. The first record it sees has no stamp, so + // the next-activation obligation (#6056) has nothing to do. + var seeded = await Mesh.GetMeshNodeStream(typePath) + .Should().Within(TestTimeouts.CrossSilo) + .Match(n => n.ContentAs(Mesh.JsonSerializerOptions) is { LastCompiledVersion: 736 } d + && d.UnreleasedBuildPath is null, + "the owner serves the settled, unstamped state"); + + var build = LateReleaseAdoption.RecordedBuild(definition)!; + var stamped = $"{typePath}/{GraphNodeTypeNames.ReleaseSegment}/20261022212254-{NodeTypeBuildState.ContentHashOf(build)}"; + return (build, stamped, seeded); + } + + /// + /// The settle's re-cut, through the production create path, while the store refuses it β€” and the + /// settle's terminal stamp, through the production field-set (ApplyCompileSuccess). + /// + private async Task RecutRefusedAndStampAsync( + string typePath, NodeCompilationResult build, string stamped, MeshNode seeded, string? classOverride = null) + { + var ct = TestContext.Current.CancellationToken; + var logger = Mesh.ServiceProvider.GetRequiredService().CreateLogger("MeshWeaver.Graph.CompileWatcher"); + var access = Mesh.ServiceProvider.GetRequiredService(); + var recut = await access + .RunAsSystem(() => NodeTypeBuildState.TryCreateReleaseNode( + Mesh, typePath, build, seeded, activityPath: null, logger, reusePath: stamped)) + .Take(1) + .Should().Within(TestTimeouts.Convergence) + .Emit("the re-cut answers", cancellationToken: ct); + Output.WriteLine($"re-cut: path={recut.ReleasePath} class={recut.FailureClass} because={recut.Because}"); + + recut.Succeeded.Should().BeFalse("the store refused the write"); + recut.AttemptedPath.Should().Be(stamped, "the re-cut names the id it was minting"); + recut.FailureClass.Should().Be(ReleaseFailureClass.Unavailable, + "a create the store never saw is classified Unavailable AT THE ARM, from the typed rejection " + + "reason β€” never parsed out of the sentence"); + + await Mesh.GetMeshNodeStream(typePath).Update(current => + { + var def = current.ContentAs(Mesh.JsonSerializerOptions)!; + return current with + { + Content = NodeTypeCompilationHelpers.ApplyCompileSuccess( + def, build, current.Version, activityPath: null, releasePath: null, + modulesHash: null, unreleasedBuildPath: recut.AttemptedPath, + unreleasedBuildReason: recut.Failure, + unreleasedBuildFailureClass: classOverride ?? recut.FailureClass), + }; + }) + .Should().Within(TestTimeouts.WriteConvergence) + .Emit("the settle's terminal stamp lands", cancellationToken: ct); + return recut; + } + + private void TheStoreAnswersAgain() + => Storage.DeliverChangeFeedGap(new ChangeFeedGap( + "postgresql:mesh_node_changes", + DateTimeOffset.UtcNow.AddSeconds(-40), + DateTimeOffset.UtcNow, + "the managed database restarted (#7068)")); + + /// + /// 🚨 THE CASE: refused as Unavailable, stamped, the store comes back, and the SAME activation + /// creates the release at the stamped id and adopts it. + /// + [Fact(Timeout = 120_000)] + public async Task AnUnavailableRecut_IsReissuedOnTheSameActivation_WhenTheStoreAnswers() + { + var ct = TestContext.Current.CancellationToken; + var typePath = $"{TestPartition}/PrWatch{Guid.NewGuid().ToString("N")[..8]}"; + var (build, stamped, seeded) = await SeedActivatedAsync(typePath); + + var refusal = Storage.RefuseWritesAsUnreachable(stamped); + try + { + await RecutRefusedAndStampAsync(typePath, build, stamped, seeded); + await refusal.Arrivals.Should().Within(TestTimeouts.Convergence) + .Emit("the re-cut's write reached the unreachable store", ct); + + var standing = await Mesh.GetMeshNodeStream(typePath) + .Should().Within(TestTimeouts.WriteConvergence) + .Match(n => n.ContentAs(Mesh.JsonSerializerOptions)?.UnreleasedBuildPath == stamped, + "the build with no release is stamped on the node"); + standing.ContentAs(Mesh.JsonSerializerOptions)!.UnreleasedBuildFailureClass + .Should().Be(ReleaseFailureClass.Unavailable, "the class travels with the stamp"); + + // Negative control 1: while the store still refuses, nothing lands β€” the pointer stays + // on the earlier build's release. + await Mesh.GetMeshNodeStream(typePath) + .Where(n => n.ContentAs(Mesh.JsonSerializerOptions)?.LatestReleasePath == stamped) + .Should().NotEmit(TestTimeouts.Quick, "the store has not come back yet"); + } + finally + { + // The store comes back. + refusal.Release(); + } + + TheStoreAnswersAgain(); + + var adopted = await Mesh.GetMeshNodeStream(typePath) + .Should().Within(TestTimeouts.CrossSilo) + .Match(n => n.ContentAs(Mesh.JsonSerializerOptions)?.LatestReleasePath == stamped, + "the activation that holds an Unavailable stamp must re-issue the create at the stamped " + + "id once the store answers again, and adopt it β€” nothing else on this activation ever " + + "creates that node (#7072)"); + var def = adopted.ContentAs(Mesh.JsonSerializerOptions)!; + def.UnreleasedBuildPath.Should().BeNull("the build has a release now"); + def.UnreleasedBuildReason.Should().BeNull("the reason never outlives the path"); + def.UnreleasedBuildFailureClass.Should().BeNull("the class never outlives the path"); + def.LastCompiledVersion.Should().Be(736L, "the re-issue recompiles nothing"); + + var release = await Mesh.GetMeshNodeStream(stamped) + .Should().Within(TestTimeouts.Convergence) + .Match(n => n.ContentAs(Mesh.JsonSerializerOptions) is not null, + "the release node exists at the ORIGINAL stamped id β€” not a fresh one"); + release.ContentAs(Mesh.JsonSerializerOptions)!.AssemblyStoreVersion + .Should().Be(736L, "the release names the recorded build"); + } + + /// + /// Negative control 2: a stamp the store DID judge (class ) + /// is a named failure, never re-issued β€” the same refusal and the same recovery signals as the + /// case above, and the stamp stands. + /// + [Fact(Timeout = 120_000)] + public async Task ARefusedRecut_IsNeverReissued_HoweverOftenTheStoreAnswers() + { + var typePath = $"{TestPartition}/TriageItem{Guid.NewGuid().ToString("N")[..8]}"; + var (build, stamped, seeded) = await SeedActivatedAsync(typePath); + + using (Storage.RefuseWritesAsUnreachable(stamped)) + await RecutRefusedAndStampAsync(typePath, build, stamped, seeded, ReleaseFailureClass.Refused); + + await Mesh.GetMeshNodeStream(typePath) + .Should().Within(TestTimeouts.WriteConvergence) + .Match(n => n.ContentAs(Mesh.JsonSerializerOptions) is { } d + && d.UnreleasedBuildPath == stamped + && d.UnreleasedBuildFailureClass == ReleaseFailureClass.Refused, + "the refused build is stamped as such", TestContext.Current.CancellationToken); + + TheStoreAnswersAgain(); + TheStoreAnswersAgain(); + + await Mesh.GetMeshNodeStream(typePath) + .Where(n => n.ContentAs(Mesh.JsonSerializerOptions) is { } d + && (d.UnreleasedBuildPath is null || d.LatestReleasePath == stamped)) + .Should().NotEmit(TestTimeouts.Quick, "a verdict is never re-issued β€” the stamp is the report"); + } + + // ───────────── the decision, pure ───────────── + + [Fact] + public void UnavailableObligation_NamesTheStamp_OnlyForTheUnavailableClass() + { + const string collection = "assemblies"; + const string contentPath = "T/v736-build.dll"; + var hash = NodeTypeBuildState.ContentHashOf(new NodeCompilationResult( + null, [], Collection: collection, ContentPath: contentPath)); + var stamped = $"T/Release/20261022212254-{hash}"; + var framework = NodeTypeCompilationHelpers.FrameworkVersion; + var owed = new NodeTypeDefinition + { + CompilationStatus = CompilationStatus.Ok, + LatestAssemblyCollection = collection, + LatestAssemblyPath = contentPath, + CompiledFrameworkVersion = framework, + UnreleasedBuildPath = stamped, + UnreleasedBuildFailureClass = ReleaseFailureClass.Unavailable, + }; + + LateReleaseAdoption.UnavailableObligation(owed, "T", framework).Should().Be(stamped); + foreach (var other in new[] { ReleaseFailureClass.Refused, ReleaseFailureClass.Expired, + ReleaseFailureClass.Faulted, "SomeFutureClass", null }) + LateReleaseAdoption.UnavailableObligation(owed with { UnreleasedBuildFailureClass = other }, "T", framework) + .Should().BeNull($"class '{other ?? "(none)"}' is not one the store never saw"); + LateReleaseAdoption.UnavailableObligation(owed with { LatestAssemblyPath = "T/v737-build.dll" }, "T", framework) + .Should().BeNull("a stamp naming other bytes is never cut for these"); + LateReleaseAdoption.UnavailableObligation(owed with { CompilationStatus = CompilationStatus.Compiling }, "T", framework) + .Should().BeNull("a compile in flight rewrites the stamp at its own settle"); + } +} diff --git a/test/MeshWeaver.Compiler.Pipeline.Test/MeshWeaver.Compiler.Pipeline.Test.csproj b/test/MeshWeaver.Compiler.Pipeline.Test/MeshWeaver.Compiler.Pipeline.Test.csproj index d378c337d7..09a6a3cd48 100644 --- a/test/MeshWeaver.Compiler.Pipeline.Test/MeshWeaver.Compiler.Pipeline.Test.csproj +++ b/test/MeshWeaver.Compiler.Pipeline.Test/MeshWeaver.Compiler.Pipeline.Test.csproj @@ -44,6 +44,9 @@ + + diff --git a/test/MeshWeaver.Compiler.Pipeline.Test/ReleaseRecutReportsWhyItFailedTest.cs b/test/MeshWeaver.Compiler.Pipeline.Test/ReleaseRecutReportsWhyItFailedTest.cs index 5d1deaa135..1a204a5836 100644 --- a/test/MeshWeaver.Compiler.Pipeline.Test/ReleaseRecutReportsWhyItFailedTest.cs +++ b/test/MeshWeaver.Compiler.Pipeline.Test/ReleaseRecutReportsWhyItFailedTest.cs @@ -3,6 +3,7 @@ using System.Reactive; using System.Reactive.Concurrency; using System.Reactive.Subjects; +using MeshWeaver.Graph.Configuration; using MeshWeaver.Mesh; using Xunit; @@ -133,6 +134,44 @@ public void AnyOtherRefusal_FailsWithItsReason_ThroughTheRealChain() Assert.Contains("cross-hub write PARTIALLY refused", outcome.Failure!); } + // ───────────── the CLASS travels with the reason (#7072) ───────────── + + /// + /// 🚨 #7072 β€” a create answered Unavailable (the store could not be reached; nothing was + /// written) is classified as such AT THE ARM, from the typed reason the create surface stamps on + /// the exception. That class is the only one the activation holding the stamp re-issues. + /// + [Fact] + public void AnUnavailableAnswer_IsClassifiedUnavailable_ThroughTheRealChain() + { + var outcome = Run((create, _) => create.OnError( + CreateNodeResponse.Fail("the data store was unreachable", NodeCreationRejectionReason.Unavailable) + .ToException(ReleasePath))); + + Assert.NotNull(outcome); + Assert.False(outcome!.Succeeded); + Assert.Equal(ReleaseFailureClass.Unavailable, outcome.FailureClass); + Assert.True(outcome.WasUnavailable); + } + + /// The other side: a VERDICT is Refused, an expiry is Expired, and neither is re-issued. + [Fact] + public void AVerdictAndAnExpiry_AreNotUnavailable_ThroughTheRealChain() + { + var refused = Run((create, _) => create.OnError( + CreateNodeResponse.Fail("Access denied", NodeCreationRejectionReason.ValidationFailed) + .ToException(ReleasePath))); + Assert.Equal(ReleaseFailureClass.Refused, refused!.FailureClass); + Assert.False(refused.WasUnavailable); + + var expired = Run((_, clock) => clock.AdvanceBy(NodeTypeBuildState.CreateBound.Add(TimeSpan.FromTicks(1)))); + Assert.Equal(ReleaseFailureClass.Expired, expired!.FailureClass); + Assert.False(expired.WasUnavailable); + + var faulted = Run((create, _) => create.OnError(new InvalidOperationException("something else"))); + Assert.Equal(ReleaseFailureClass.Faulted, faulted!.FailureClass); + } + // ───────────────── the channel the incident names: a bound that expired ───────────────── /// diff --git a/test/MeshWeaver.Testing.FaultInjection/FaultInjectingStorageAdapter.cs b/test/MeshWeaver.Testing.FaultInjection/FaultInjectingStorageAdapter.cs index 82fe3c5a12..d36cfcb016 100644 --- a/test/MeshWeaver.Testing.FaultInjection/FaultInjectingStorageAdapter.cs +++ b/test/MeshWeaver.Testing.FaultInjection/FaultInjectingStorageAdapter.cs @@ -50,6 +50,7 @@ public sealed class FaultInjectingStorageAdapter : IStorageAdapter private readonly ConcurrentDictionary _conditionalWriteFaults = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary _deleteManyFaults = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary _deleteFaults = new(StringComparer.OrdinalIgnoreCase); + private readonly ConcurrentDictionary _unreachableWrites = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary> _conditionalDeleteResults = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary> _heldWrites = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary> _enumerations = new(StringComparer.OrdinalIgnoreCase); @@ -122,6 +123,23 @@ public FaultSwitch FailConditionalWrites(string path) s => _conditionalWriteFaults.TryRemove(new KeyValuePair(k, s)))); } + /// + /// From now on, every write of (Write, WriteIfVersion, a + /// WriteMany naming it) FAULTS with a driver-shaped CONNECTION failure β€” a + /// with SQLSTATE 08006, which is what a + /// managed PostgreSQL restart answers (MeshWeaver#7068) β€” and writes nothing. The create + /// handler classifies it as an unreachable store and answers Unavailable, exactly as + /// production did (MeshWeaver#7072). Releasing the switch is the store coming back. Each refused + /// write is an arrival on the switch. + /// + /// The node path whose writes cannot reach the store. + public FaultSwitch RefuseWritesAsUnreachable(string path) + { + var key = Norm(path); + return _unreachableWrites.GetOrAdd(key, k => new FaultSwitch($"{Name}: store unreachable for writes of {k}", + s => _unreachableWrites.TryRemove(new KeyValuePair(k, s)))); + } + /// /// From now on, a DeleteMany whose batch names deletes every OTHER /// path of the batch and then FAULTS, leaving stored β€” a batch delete that @@ -269,7 +287,25 @@ public IObservable ReadMany(IReadOnlyCollection paths, JsonSer /// public IObservable Write(MeshNode node, JsonSerializerOptions options) - => HeldWrite(node, () => _inner.Write(node, options)); + => Observable.Defer(() => Unreachable(node.Path, $"Write {node.Path}") + ?? HeldWrite(node, () => _inner.Write(node, options))); + + /// The refusal a closed answers for + /// , or null when the store is reachable for it. + private IObservable? Unreachable(string path, string what) + { + if (!_unreachableWrites.TryGetValue(Norm(path), out var fault) || !fault.IsClosed) + return null; + fault.NoteArrival(what); + return Observable.Throw(new UnreachableStoreException( + $"{Name}: injected fault β€” failed to connect to the store for {path}")); + } + + /// A driver connection failure (SQLSTATE 08006, connection_failure). + private sealed class UnreachableStoreException(string message) : System.Data.Common.DbException(message) + { + public override string? SqlState => "08006"; + } /// public IObservable> WriteMany(IReadOnlyCollection nodes, JsonSerializerOptions options) @@ -278,6 +314,9 @@ public IObservable> WriteMany(IReadOnlyCollection Unreachable>(n.Path, $"WriteMany {n.Path}")) + .FirstOrDefault(r => r is not null) is { } refused) + return refused; var held = nodes.Select(n => (Node: n, Hold: WriteHold(n.Path))) .Where(h => h.Hold is not null).ToList(); if (held.Count == 0 || held[0].Hold is not { } gate) @@ -293,6 +332,8 @@ public IObservable> WriteMany(IReadOnlyCollection WriteIfVersion(MeshNode node, long expectedVersion, JsonSerializerOptions options) => Observable.Defer(() => { + if (Unreachable(node.Path, $"WriteIfVersion {node.Path}") is { } refused) + return refused; if (_conditionalWriteFaults.TryGetValue(Norm(node.Path), out var fault) && fault.IsClosed) { fault.NoteArrival($"WriteIfVersion {node.Path} v{node.Version}"); From 2c4788aac2966bccaf217428e30f95e8bb4566a6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= <6334612+rbuergi@users.noreply.github.com> Date: Fri, 23 Oct 2026 15:56:03 +0200 Subject: [PATCH 2/7] fix(release): keep re-issue evidence across an attempt, weigh it by source, record a re-issue's verdict (review on #7087) - ReissueWhenStoreAnswers observes evidence for its whole life: evidence arriving during an in-flight attempt is retained and starts the next one after an Unavailable answer (gaps are not replayed, so it was lost). - Each attempt remembers whether a commit or a gap started it; only a commit-started Unavailable answer disproves commits as evidence. - A re-issue answered with a verdict writes its class and reason to the record through the owner stream, only while the same Unavailable obligation stands, so a rebuilt watch never re-issues a judged create. - Injector RefuseWritesAsJudged; deterministic regressions for each. StoreAnswersAgain keeps its public signature. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../LateReleaseAdoption.cs | 57 ++++- .../Architecture/ReleaseFailureClasses.md | 28 ++- .../StoreReachability.cs | 145 ++++++++++-- ...eRecutIsReissuedWhenTheStoreAnswersTest.cs | 213 ++++++++++++++++++ .../FaultInjectingStorageAdapter.cs | 27 ++- 5 files changed, 436 insertions(+), 34 deletions(-) diff --git a/src/MeshWeaver.Compiler.Pipeline/LateReleaseAdoption.cs b/src/MeshWeaver.Compiler.Pipeline/LateReleaseAdoption.cs index 2c82806f30..0fb2e0ff4a 100644 --- a/src/MeshWeaver.Compiler.Pipeline/LateReleaseAdoption.cs +++ b/src/MeshWeaver.Compiler.Pipeline/LateReleaseAdoption.cs @@ -287,6 +287,39 @@ internal static IObservable CompleteInheritedRelease( ? InheritedObligation(definition, hubPath, frameworkVersion) : null; + /// + /// The record after a re-issue ended in a VERDICT rather than another Unavailable answer (review on + /// #7087), or null to leave it alone. Pure. + /// + /// The stamp keeps its path β€” this build still has no release β€” but its reason and class + /// become the re-issue's own, so the record says what the store answered last and + /// no longer names it: a watch rebuilt from this record does not + /// re-issue a create the store has judged. Applied only while the record still holds the SAME + /// unavailable obligation; a later settle, an adoption or another writer that moved the stamp on + /// owns it now. + /// + /// The definition as the owner currently holds it. + /// The NodeType's path. + /// The stamped path the re-issue was for. + /// The re-issue's outcome β€” a failure that was not Unavailable. + internal static NodeTypeDefinition? RecordReissueVerdict( + NodeTypeDefinition? definition, string hubPath, string stamped, + NodeTypeBuildState.ReleaseCreateOutcome verdict) + { + if (verdict.Succeeded || verdict.WasUnavailable + || !string.Equals( + UnavailableObligation(definition, hubPath, NodeTypeCompilationHelpers.FrameworkVersion), + stamped, StringComparison.Ordinal)) + return null; + return definition! with + { + UnreleasedBuildReason = verdict.Failure is { Length: > 0 } reason + ? reason + : "the re-issue reported no reason, which is itself a defect in this pipeline", + UnreleasedBuildFailureClass = verdict.FailureClass ?? ReleaseFailureClass.Faulted, + }; + } + /// /// #7072 β€” the activation's half of : while the own record /// carries such a stamp, re-issue the release create at the stamped id once the store is @@ -351,14 +384,28 @@ internal static IObservable ReissueWhenTheStoreAnswers( { if (outcome.ReleasePath is not { } landed) { - // A verdict, not an outage: the stamp stands and is the report; the next - // activation cuts it once more (#6056). + // A verdict, not an outage. The stamp stands and is the report β€” so it + // must report THIS verdict (review on #7087): left at Unavailable with the + // old reason, it would misstate the failure, and a rebuilt watch (a + // transient fault on the own stream re-establishes it from the record) + // would re-issue a create the store has now judged. The new class and + // reason go to the record through the owner, and only while the record + // still holds the SAME unavailable obligation; the next activation cuts + // the stamp once more (#6056). logger?.LogWarning( "[ReleasePostCondition] {HubPath}: re-issuing the release at " + "{ReleasePath} after the store answered again did not land{Because}. " - + "The stamp stands (#7072).", - hubPath, stamped, outcome.Because); - return Observable.Empty(); + + "The stamp stands, now with this verdict's class {FailureClass} (#7072).", + hubPath, stamped, outcome.Because, + outcome.FailureClass ?? ReleaseFailureClass.Faulted); + return accessService.RunAsSystem(() => workspace.GetMeshNodeStream() + .Update(current => + RecordReissueVerdict(current.ContentAs(options), + hubPath, stamped, outcome) is { } recorded + ? current with { Content = recorded } + : current)) + .Take(1) + .SelectMany(_ => Observable.Empty()); } return accessService.RunAsSystem(() => workspace.GetMeshNodeStream() .Update(current => diff --git a/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md b/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md index f9fb297e82..141a499aac 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md @@ -577,10 +577,21 @@ The log line said the retry was meaningful, and no component on the live pod mad `MeshWeaver.Mesh.Contract` waits for `StoreAnswersAgain`. That signal fires on a declared `ChangeFeedGap`, which is the LISTEN connection resuming after a database restart, or on a commit announced in the type's partition, which the store must have accepted. While the store is away, - neither fires and nothing is re-issued. If a re-issue is answered `Unavailable` again after a - partition commit, that commit has been shown not to be evidence for this write. Later re-issues - then wait for a feed reconnect only. A busy partition therefore cannot turn this into a retry - loop, and no count or clock is involved. + neither fires and nothing is re-issued. Evidence is observed for the whole life of the watch: + evidence that arrives while an attempt is in flight is kept (the event that started the attempt + is not), and starts the next attempt as soon as that one is answered `Unavailable`. A feed does + not replay its gaps, so a reconnect seen only between attempts would be lost. Each attempt + remembers which kind of evidence started it. If an attempt STARTED BY A COMMIT is answered + `Unavailable` again, commits have been shown not to be evidence for this write, and from then on + only a feed reconnect starts a re-issue. A failed attempt started by a reconnect disproves nothing + about commits. A busy partition therefore cannot turn this into a retry loop, and no count or + clock is involved. +4. **A verdict on a re-issue is written back.** If the store comes back and answers the re-issue + with a verdict (`Refused`, `Faulted`), `RecordReissueVerdict` replaces the stamp's reason and + class with that verdict's, through the owner stream, and only while the record still holds the + SAME `Unavailable` obligation. The stamp keeps its path. A watch rebuilt from the record (a + transient own-stream fault re-establishes it) then has no obligation and never re-issues a create + the store has judged. Every other class stays standing. A `Refused` create would be refused again. An `Expired` one may still land, and late adoption takes it when it does. A stamp written before the class existed has @@ -614,6 +625,15 @@ stamp under the same signals is never re-issued. With the re-issue watch removed the main case is red at its final wait. `ReleaseRecutReportsWhyItFailedTest` pins the classification through the real `Bounded` chain. +The evidence handling is pinned deterministically against the real `InProcessMeshChangeFeed`, with +a write the test answers by hand. `AGapDuringAnAttempt_StartsTheNextAttempt_WhenThatOneIsUnavailable` +covers a reconnect that lands during an attempt. `AGapStartedFailure_KeepsCommitsAsEvidence` and +`ACommitStartedFailure_LeavesOnlyReconnectsAsEvidence` cover which source a failure disproves. +`AReissueTheStoreRefuses_RecordsTheRefusal_AndIsNotReissuedAgain` (`RefuseWritesAsJudged`) covers +the written-back verdict. With the first shape of `ReissueWhenStoreAnswers` (evidence observed only +between attempts, any failure switching to reconnect-only) the first two are red. With +`RecordReissueVerdict` disabled, the last is red. + ### Not established The memex pod's own `StoreAnswersAgain` evidence for that night was not read. Whether its LISTEN diff --git a/src/MeshWeaver.Mesh.Contract/StoreReachability.cs b/src/MeshWeaver.Mesh.Contract/StoreReachability.cs index fbe91fe762..79a79f0863 100644 --- a/src/MeshWeaver.Mesh.Contract/StoreReachability.cs +++ b/src/MeshWeaver.Mesh.Contract/StoreReachability.cs @@ -1,5 +1,7 @@ ο»Ώusing System.Reactive; +using System.Reactive.Disposables; using System.Reactive.Linq; +using System.Reactive.Subjects; using MeshWeaver.Data; using MeshWeaver.Mesh.Services; @@ -122,15 +124,24 @@ public static string DescribeMayHavePartiallyLanded(string operation) /// null to count every commit. /// An infinite signal; one element per piece of evidence. public static IObservable StoreAnswersAgain(this IMeshInvalidationFeed? feed, string? partition) + => EvidenceBySource(feed, partition).Select(_ => Unit.Default); + + /// + /// with its SOURCE kept: true for a declared gap (the feed + /// resumed), false for a commit in the partition. The two are different kinds of evidence β€” + /// a commit can be shown not to count for one particular write, a reconnect cannot β€” so a caller + /// that weighs them must be able to tell which fired. + /// + private static IObservable EvidenceBySource(IMeshInvalidationFeed? feed, string? partition) { if (feed is null) - return Observable.Never(); + return Observable.Never(); // The handler is forwarded as-is and the filter runs as an operator, so a fault in it reaches // the subscriber's OnError instead of being thrown out of the feed's delivery callback. var commits = Observable.Create(observer => feed.Subscribe(observer.OnNext)) .Where(change => partition is null || InPartition(change.Path, partition)) - .Select(_ => Unit.Default); - return commits.Merge(feed.Gaps.Select(_ => Unit.Default)); + .Select(_ => false); + return commits.Merge(feed.Gaps.Select(_ => true)); } /// @@ -139,17 +150,25 @@ public static IObservable StoreAnswersAgain(this IMeshInvalidationFeed? fe /// The ONE shape for every such site: the caller supplies the SAME request (same node id), how /// to tell an Unavailable outcome, and the partition the write lands in. /// - /// How it waits. The first re-issue waits for β€” - /// a feed that resumed, or a commit in the partition. If THAT re-issue is answered Unavailable - /// again, a commit in the partition has just been shown not to be evidence for this write (the - /// store took someone else's commit and still never saw ours), so every later re-issue waits for - /// the feed to RESUME instead β€” a reconnect, which is rare. That is what keeps this from turning a - /// busy partition into a retry loop, without any count or clock. + /// How it waits. A re-issue starts on the first piece of + /// evidence β€” a feed that resumed, or a commit in the partition. + /// The evidence is observed for the WHOLE life of the sequence, never only between attempts: + /// evidence that arrives while an attempt is in flight is RETAINED (the event that started the + /// attempt is not), and if the attempt is answered Unavailable it starts the next one at once. A + /// feed's gaps are not replayed, so a reconnect that landed during the attempt would otherwise be + /// lost for good. + /// + /// Which evidence still counts. A re-issue STARTED BY A COMMIT that is answered + /// Unavailable again shows that a commit in the partition is not evidence for this write (the + /// store took someone else's commit and still never saw ours), so from then on only a reconnect + /// starts a re-issue. A re-issue started by a reconnect that fails disproves nothing about commits, + /// and commits keep counting. That is what keeps a busy partition from becoming a retry loop, + /// without any count or clock. /// /// What it never does. It never re-issues a write that was JUDGED: an outcome /// rejects is emitted and the sequence completes, so a refusal - /// stays a named failure. And it never issues the FIRST attempt β€” the caller made it, and is here - /// because it came back Unavailable. + /// stays a named failure. It never runs two attempts at once. And it never issues the FIRST + /// attempt β€” the caller made it, and is here because it came back Unavailable. /// /// The write's outcome. /// The mesh's invalidation feed, or null (then nothing is re-issued). @@ -167,19 +186,99 @@ public static IObservable ReissueWhenStoreAnswers( { if (feed is null) return Observable.Never(); - var reconnects = feed.Gaps.Select(_ => Unit.Default); - IObservable After(IObservable evidence) => - evidence - .Take(1) - .SelectMany(_ => Observable.Defer(attempt).Take(1)) - .SelectMany(outcome => + return Observable.Create(observer => + { + // Every transition runs inside the ONE merged subscription below, which Rx serialises, so + // this state is touched by one callback at a time. An attempt that answers synchronously + // re-enters that callback; every transition therefore finishes its state change BEFORE it + // starts an attempt, so a re-entrant call always sees a consistent state. + var state = new ReissueState(); + var outcomes = new Subject<(T Outcome, bool ByGap)>(); + var inFlight = new SerialDisposable(); + + void Start(bool byGap) + { + state.InFlight = true; + state.PendingGap = false; + state.PendingCommit = false; + var slot = new SingleAssignmentDisposable(); + inFlight.Disposable = slot; + var answered = false; + slot.Disposable = Observable.Defer(attempt).Take(1).Subscribe( + outcome => + { + answered = true; + outcomes.OnNext((outcome, byGap)); + }, + observer.OnError, + () => + { + if (!answered) + observer.OnError(new InvalidOperationException( + "A re-issued write completed without an outcome, so whether the store " + + "saw it is unknown β€” refusing to wait on evidence for an answer that " + + "will never come.")); + }); + } + + var events = EvidenceBySource(feed, partition) + .Select(byGap => (IsOutcome: false, ByGap: byGap, Outcome: default(T))) + .Merge(outcomes.Select(x => (IsOutcome: true, x.ByGap, Outcome: (T?)x.Outcome))); + + var subscription = events.Subscribe(e => + { + if (state.Done) + return; + if (!e.IsOutcome) { - if (!wasUnavailable(outcome)) - return Observable.Return(outcome); - onStillUnavailable?.Invoke(outcome); - return Observable.Defer(() => After(reconnects)); - }); - return Observable.Defer(() => After(feed.StoreAnswersAgain(partition))); + if (!e.ByGap && !state.CommitsCount) + return; + if (state.InFlight) + { + // Retained: newer than the event that started the attempt in flight. + if (e.ByGap) state.PendingGap = true; + else state.PendingCommit = true; + return; + } + Start(e.ByGap); + return; + } + + state.InFlight = false; + var outcome = e.Outcome!; + if (!wasUnavailable(outcome)) + { + state.Done = true; + observer.OnNext(outcome); + observer.OnCompleted(); + return; + } + onStillUnavailable?.Invoke(outcome); + if (!e.ByGap) + { + // A commit started it and the store still never saw it: commits are disproved. + state.CommitsCount = false; + state.PendingCommit = false; + } + if (state.PendingGap) + Start(byGap: true); + else if (state.PendingCommit) + Start(byGap: false); + }, observer.OnError); + + return new CompositeDisposable(subscription, inFlight, outcomes); + }); + } + + /// The state of one subscription β€” owned by + /// that subscription alone and touched only from its serialised callback. + private sealed class ReissueState + { + public bool InFlight; + public bool CommitsCount = true; + public bool PendingGap; + public bool PendingCommit; + public bool Done; } private static bool InPartition(string? path, string partition) diff --git a/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs b/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs index e00b83fd50..573b5e2578 100644 --- a/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs +++ b/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs @@ -7,6 +7,8 @@ using MeshWeaver.Fixture; using MeshWeaver.Graph; using MeshWeaver.Graph.Configuration; +using MeshWeaver.Hosting; +using MeshWeaver.Hosting.Persistence; using MeshWeaver.Hosting.Monolith.TestBase; using MeshWeaver.Mesh; using MeshWeaver.Mesh.Services; @@ -240,6 +242,217 @@ await Mesh.GetMeshNodeStream(typePath) .Should().NotEmit(TestTimeouts.Quick, "a verdict is never re-issued β€” the stamp is the report"); } + /// + /// Review on #7087, thread 3: a re-issue that the store answers with a VERDICT must leave that + /// verdict on the record. The stamp is Unavailable; the store comes back and REFUSES the re-issue. + /// The record must then carry the refusal's class and reason, so a watch rebuilt from it (a + /// transient own-stream fault re-establishes it) has no Unavailable obligation and never re-issues + /// a create the store has judged. Without the owner write the record keeps Unavailable and + /// the old reason, and this case is red at its first Match. + /// + [Fact(Timeout = 120_000)] + public async Task AReissueTheStoreRefuses_RecordsTheRefusal_AndIsNotReissuedAgain() + { + var ct = TestContext.Current.CancellationToken; + var typePath = $"{TestPartition}/TriageStatus{Guid.NewGuid().ToString("N")[..8]}"; + var (build, stamped, seeded) = await SeedActivatedAsync(typePath); + + using (Storage.RefuseWritesAsUnreachable(stamped)) + await RecutRefusedAndStampAsync(typePath, build, stamped, seeded); + + var judged = Storage.RefuseWritesAsJudged(stamped); + try + { + TheStoreAnswersAgain(); + await judged.Arrivals.Should().Within(TestTimeouts.Convergence) + .Emit("the store came back and the re-issue reached it", ct); + + var recorded = await Mesh.GetMeshNodeStream(typePath) + .Should().Within(TestTimeouts.WriteConvergence) + .Match(n => n.ContentAs(Mesh.JsonSerializerOptions) is { } d + && d.UnreleasedBuildFailureClass == ReleaseFailureClass.Refused, + "the re-issue's verdict replaces the stale Unavailable class on the record", ct); + var def = recorded.ContentAs(Mesh.JsonSerializerOptions)!; + def.UnreleasedBuildPath.Should().Be(stamped, "this build still has no release"); + def.UnreleasedBuildReason.Should().Contain("the store refused the write", + "the reason is the verdict's own, not the earlier outage's"); + LateReleaseAdoption.UnavailableObligation(def, typePath, NodeTypeCompilationHelpers.FrameworkVersion) + .Should().BeNull("a watch rebuilt from this record has nothing to re-issue"); + } + finally + { + judged.Release(); + } + + // Watcher recovery: more evidence, the store now taking writes; a judged stamp is never re-issued. + TheStoreAnswersAgain(); + await Mesh.GetMeshNodeStream(typePath) + .Where(n => n.ContentAs(Mesh.JsonSerializerOptions) is { } d + && (d.LatestReleasePath == stamped || d.UnreleasedBuildPath is null)) + .Should().NotEmit(TestTimeouts.Quick, "a create the store judged is not re-issued on this activation"); + } + + // ───────────── the re-issue's evidence handling, deterministic (review on #7087) ───────────── + // + // Driven against the REAL InProcessMeshChangeFeed over the fault-injecting store, with a write + // the test answers by hand, so every ordering is chosen rather than raced. + + private const string Unavailable = "unavailable"; + private const string Landed = "landed"; + + private sealed class Rig : IDisposable + { + private readonly IDisposable _subscription; + + public Rig() + { + Feed = new InProcessMeshChangeFeed(Store); + _subscription = Feed + .ReissueWhenStoreAnswers( + "P", + () => + { + var attempt = new System.Reactive.Subjects.Subject(); + Attempts = Attempts.Add(attempt); + return attempt; + }, + outcome => outcome == Unavailable, + outcome => StillUnavailable = StillUnavailable.Add(outcome)) + .Subscribe(r => Results = Results.Add(r)); + } + + public FaultInjectingStorageAdapter Store { get; } = new(new InMemoryStorageAdapter()); + public InProcessMeshChangeFeed Feed { get; } + public ImmutableList> Attempts { get; private set; } = []; + public ImmutableList Results { get; private set; } = []; + public ImmutableList StillUnavailable { get; private set; } = []; + + public void Commit() => Feed.Publish(new MeshChangeEvent( + "P", "other", "P/other", MeshChangeKind.Updated, null, 1, DateTimeOffset.UtcNow)); + + public void Gap() => Store.DeliverChangeFeedGap(new ChangeFeedGap( + "postgresql:mesh_node_changes", DateTimeOffset.UtcNow.AddSeconds(-1), DateTimeOffset.UtcNow)); + + public void Answer(int attempt, string outcome) => Attempts[attempt].OnNext(outcome); + + public void Dispose() + { + _subscription.Dispose(); + Feed.Dispose(); + } + } + + /// + /// Thread 1: a reconnect that lands WHILE an attempt is in flight is retained, and starts the next + /// attempt the moment that one is answered Unavailable. A feed's gaps are not replayed, so the old + /// shape, which looked for evidence only after the failure, lost it for good. + /// + [Fact] + public void AGapDuringAnAttempt_StartsTheNextAttempt_WhenThatOneIsUnavailable() + { + using var rig = new Rig(); + rig.Commit(); + rig.Attempts.Should().HaveCount(1, "a commit in the partition is evidence"); + rig.Gap(); + rig.Attempts.Should().HaveCount(1, "never two attempts at once"); + + rig.Answer(0, Unavailable); + rig.Attempts.Should().HaveCount(2, "the gap that arrived during the attempt is retained, not lost"); + rig.Answer(1, Landed); + rig.Results.Should().Equal(Landed); + } + + /// + /// Thread 2: a failed attempt that a GAP started disproves nothing about commits, so a later commit + /// in the partition still starts a re-issue. The old shape went reconnect-only after ANY failure. + /// + [Fact] + public void AGapStartedFailure_KeepsCommitsAsEvidence() + { + using var rig = new Rig(); + rig.Gap(); + rig.Answer(0, Unavailable); + rig.Attempts.Should().HaveCount(1); + + rig.Commit(); + rig.Attempts.Should().HaveCount(2, "a commit is still evidence; only a commit-started failure disproves it"); + rig.Answer(1, Landed); + rig.Results.Should().Equal(Landed); + } + + /// + /// The other side: a failed attempt that a COMMIT started shows commits are not evidence for this + /// write. Later commits start nothing (no retry loop on a busy partition), a reconnect does, and a + /// commit retained during that attempt is discarded with the rest. + /// + [Fact] + public void ACommitStartedFailure_LeavesOnlyReconnectsAsEvidence() + { + using var rig = new Rig(); + rig.Commit(); + rig.Commit(); + rig.Answer(0, Unavailable); + rig.Attempts.Should().HaveCount(1, "the commit retained during a commit-started failure is disproved too"); + + rig.Commit(); + rig.Commit(); + rig.Attempts.Should().HaveCount(1, "commits no longer start a re-issue"); + rig.StillUnavailable.Should().HaveCount(1); + + rig.Gap(); + rig.Attempts.Should().HaveCount(2, "a reconnect always counts"); + rig.Answer(1, Landed); + rig.Results.Should().Equal(Landed); + } + + /// A verdict ends the sequence: it is emitted, and no evidence after it starts anything. + [Fact] + public void AVerdict_IsEmitted_AndNothingIsReissuedAfterIt() + { + using var rig = new Rig(); + rig.Gap(); + rig.Answer(0, "refused"); + rig.Results.Should().Equal("refused"); + rig.Gap(); + rig.Commit(); + rig.Attempts.Should().HaveCount(1); + } + + [Fact] + public void RecordReissueVerdict_ReplacesTheClass_OnlyWhileTheSameObligationStands() + { + const string collection = "assemblies"; + const string contentPath = "T/v736-build.dll"; + var hash = NodeTypeBuildState.ContentHashOf(new NodeCompilationResult( + null, [], Collection: collection, ContentPath: contentPath)); + var stamped = $"T/Release/20261022212254-{hash}"; + var owed = new NodeTypeDefinition + { + CompilationStatus = CompilationStatus.Ok, + LatestAssemblyCollection = collection, + LatestAssemblyPath = contentPath, + CompiledFrameworkVersion = NodeTypeCompilationHelpers.FrameworkVersion, + UnreleasedBuildPath = stamped, + UnreleasedBuildReason = "the data store was unreachable", + UnreleasedBuildFailureClass = ReleaseFailureClass.Unavailable, + }; + var refused = NodeTypeBuildState.ReleaseCreateOutcome.Failed("refused", stamped, ReleaseFailureClass.Refused); + + var recorded = LateReleaseAdoption.RecordReissueVerdict(owed, "T", stamped, refused)!; + recorded.UnreleasedBuildPath.Should().Be(stamped); + recorded.UnreleasedBuildReason.Should().Be("refused"); + recorded.UnreleasedBuildFailureClass.Should().Be(ReleaseFailureClass.Refused); + + LateReleaseAdoption.RecordReissueVerdict(owed, "T", stamped, + NodeTypeBuildState.ReleaseCreateOutcome.Failed("again", stamped, ReleaseFailureClass.Unavailable)) + .Should().BeNull("an Unavailable answer leaves the obligation standing"); + LateReleaseAdoption.RecordReissueVerdict(owed with { UnreleasedBuildPath = null }, "T", stamped, refused) + .Should().BeNull("a later writer moved the stamp on"); + LateReleaseAdoption.RecordReissueVerdict( + owed with { UnreleasedBuildFailureClass = ReleaseFailureClass.Expired }, "T", stamped, refused) + .Should().BeNull("only the SAME unavailable obligation is overwritten"); + } + // ───────────── the decision, pure ───────────── [Fact] diff --git a/test/MeshWeaver.Testing.FaultInjection/FaultInjectingStorageAdapter.cs b/test/MeshWeaver.Testing.FaultInjection/FaultInjectingStorageAdapter.cs index d36cfcb016..b83e6fe536 100644 --- a/test/MeshWeaver.Testing.FaultInjection/FaultInjectingStorageAdapter.cs +++ b/test/MeshWeaver.Testing.FaultInjection/FaultInjectingStorageAdapter.cs @@ -51,6 +51,7 @@ public sealed class FaultInjectingStorageAdapter : IStorageAdapter private readonly ConcurrentDictionary _deleteManyFaults = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary _deleteFaults = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary _unreachableWrites = new(StringComparer.OrdinalIgnoreCase); + private readonly ConcurrentDictionary _judgedWrites = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary> _conditionalDeleteResults = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary> _heldWrites = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary> _enumerations = new(StringComparer.OrdinalIgnoreCase); @@ -140,6 +141,21 @@ public FaultSwitch RefuseWritesAsUnreachable(string path) s => _unreachableWrites.TryRemove(new KeyValuePair(k, s)))); } + /// + /// From now on, every write of FAULTS with an + /// β€” a store that was reached and REFUSED the write, which + /// the create handler answers ValidationFailed: a verdict, not an outage. The counterpart of + /// , for a case that needs the store to judge a write it + /// earlier could not see (MeshWeaver#7087). Each refused write is an arrival on the switch. + /// + /// The node path whose writes the store refuses. + public FaultSwitch RefuseWritesAsJudged(string path) + { + var key = Norm(path); + return _judgedWrites.GetOrAdd(key, k => new FaultSwitch($"{Name}: store refuses writes of {k}", + s => _judgedWrites.TryRemove(new KeyValuePair(k, s)))); + } + /// /// From now on, a DeleteMany whose batch names deletes every OTHER /// path of the batch and then FAULTS, leaving stored β€” a batch delete that @@ -290,10 +306,17 @@ public IObservable ReadMany(IReadOnlyCollection paths, JsonSer => Observable.Defer(() => Unreachable(node.Path, $"Write {node.Path}") ?? HeldWrite(node, () => _inner.Write(node, options))); - /// The refusal a closed answers for - /// , or null when the store is reachable for it. + /// The refusal a closed or + /// answers for , or null when + /// the store takes the write. private IObservable? Unreachable(string path, string what) { + if (_judgedWrites.TryGetValue(Norm(path), out var judged) && judged.IsClosed) + { + judged.NoteArrival(what); + return Observable.Throw(new InvalidOperationException( + $"{Name}: injected fault β€” the store refused the write of {path}")); + } if (!_unreachableWrites.TryGetValue(Norm(path), out var fault) || !fault.IsClosed) return null; fault.NoteArrival(what); From 6075cad6af3637b74f3dffe4600e7067e6cda0f7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= <6334612+rbuergi@users.noreply.github.com> Date: Fri, 23 Oct 2026 20:51:32 +0200 Subject: [PATCH 3/7] ci(shards): run MeshWeaver.Compiler.Pipeline.Test in three parts (#7087) PR run 39965744393 killed MeshWeaver.Compiler.Pipeline.Test at the 480 s per-project cap (exit=124 TIMEOUT, recorded as HOST_CRASHED), still progressing through its classes: no hang, every TEST START before the cut had its TEST END except the class running at 480 s. The project measured 388 s solo on PR run 39964851393 (81% of the cap, already past the 288 s headroom threshold while listed at 227 s), and #7072's regression class adds three deliberate 36 s negative waits (NotEmit(TestTimeouts.Quick) on CI). Per the split rule in shard-assign.sh the cap is not raised; the project is split instead. ShardWeightHeadroomGuard fails on the measured weight unsplit (500 s / 1 part) and passes with three parts. Co-Authored-By: Claude Opus 5.5 (1M context) --- .github/scripts/shard-assign.sh | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/.github/scripts/shard-assign.sh b/.github/scripts/shard-assign.sh index f95534a51b..6593e59ab8 100755 --- a/.github/scripts/shard-assign.sh +++ b/.github/scripts/shard-assign.sh @@ -107,6 +107,13 @@ # 428 s on each of the three green runs and 435 s on 37830774943; it now # runs in two parts. # +# MeshWeaver.Compiler.Pipeline.Test was the one after that: listed at 227 s, it measured +# 388 s solo on PR run 39964851393 (81% of the cap, already past the 288 s threshold) and was +# killed at 480 s exit=124 on PR run 39965744393 (#7087), still progressing through its +# classes, once #7072's regression class added three 36 s deliberate negative waits +# (`NotEmit(TestTimeouts.Quick)` is 36 s on CI). Measured total ~500 s; it runs in three +# parts since, so no every-Nth parity can stack its 36 s classes past the threshold. +# # 🚨 The two triggers must BOTH be checked when re-measuring. A project that grows # past 288 s (60% of 480) needs splitting even while the LPT loop still reports # six balanced shards, which is precisely why the balance rule alone missed this. @@ -158,7 +165,7 @@ WEIGHTS=$(cat <<'EOF' 43 MeshWeaver.Layout.Test 7 MeshWeaver.ContentCollections.Test 37 MeshWeaver.Documentation.Test -227 MeshWeaver.Compiler.Pipeline.Test +500 MeshWeaver.Compiler.Pipeline.Test 3 2 MeshWeaver.ContentCollections.Indexing.Test 2 MeshWeaver.Markdown.Collaboration.Test 2 MeshWeaver.Portal.E2E.Test From 09aafec04081724b89d83b96f3a4775321529896 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= <6334612+rbuergi@users.noreply.github.com> Date: Fri, 23 Oct 2026 20:55:33 +0200 Subject: [PATCH 4/7] fix(release): connect re-issue outcomes before evidence; weigh a failure before going idle (review on #7087) Two re-entrancy defects in ReissueWhenStoreAnswers, both from Copilot review: - Evidence was the first Merge source, so evidence a feed delivered while being subscribed ran Start before the hot outcomes subject had a subscriber; an attempt answering in that window lost its outcome and the sequence stayed in flight for good. Outcomes are now connected first. - The attempt went idle before wasUnavailable/onStillUnavailable ran, so a commit those callbacks published re-entered and started a nested attempt before a commit-started failure disproved commits, clearing a retained reconnect. The attempt now stays in flight until its outcome is weighed. Three deterministic cases; each fails without the fix and passes with it. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Architecture/ReleaseFailureClasses.md | 7 +- .../StoreReachability.cs | 21 ++++- ...eRecutIsReissuedWhenTheStoreAnswersTest.cs | 89 ++++++++++++++++++- 3 files changed, 110 insertions(+), 7 deletions(-) diff --git a/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md b/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md index 141a499aac..a58e5303ae 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md @@ -585,7 +585,12 @@ The log line said the retry was meaningful, and no component on the live pod mad `Unavailable` again, commits have been shown not to be evidence for this write, and from then on only a feed reconnect starts a re-issue. A failed attempt started by a reconnect disproves nothing about commits. A busy partition therefore cannot turn this into a retry loop, and no count or - clock is involved. + clock is involved. Two orderings keep that true under synchronous re-entry. The watch connects + its own outcome stream BEFORE the evidence feed, so evidence a feed delivers while it is being + subscribed cannot start an attempt whose answer has nowhere to go. And a failed attempt stays in + flight until its outcome has been weighed, so a commit published from inside the + still-unavailable callback is retained and judged like any other evidence. It never starts a + nested attempt that would clear a retained reconnect. 4. **A verdict on a re-issue is written back.** If the store comes back and answers the re-issue with a verdict (`Refused`, `Faulted`), `RecordReissueVerdict` replaces the stamp's reason and class with that verdict's, through the owner stream, and only while the record still holds the diff --git a/src/MeshWeaver.Mesh.Contract/StoreReachability.cs b/src/MeshWeaver.Mesh.Contract/StoreReachability.cs index 79a79f0863..2ef8f3b824 100644 --- a/src/MeshWeaver.Mesh.Contract/StoreReachability.cs +++ b/src/MeshWeaver.Mesh.Contract/StoreReachability.cs @@ -221,9 +221,14 @@ void Start(bool byGap) }); } - var events = EvidenceBySource(feed, partition) - .Select(byGap => (IsOutcome: false, ByGap: byGap, Outcome: default(T))) - .Merge(outcomes.Select(x => (IsOutcome: true, x.ByGap, Outcome: (T?)x.Outcome))); + // 🚨 The OUTCOMES are connected FIRST (review on #7087). Merge subscribes its sources in + // order, and a feed may deliver evidence while it is being subscribed; with the evidence + // first, that evidence ran Start before the hot `outcomes` subject had a subscriber, an + // attempt that answered in that window lost its outcome, and InFlight stayed true for good + // β€” every later piece of evidence was retained and none ever started an attempt. + var events = outcomes.Select(x => (IsOutcome: true, x.ByGap, Outcome: (T?)x.Outcome)) + .Merge(EvidenceBySource(feed, partition) + .Select(byGap => (IsOutcome: false, ByGap: byGap, Outcome: default(T)))); var subscription = events.Subscribe(e => { @@ -244,11 +249,18 @@ void Start(bool byGap) return; } - state.InFlight = false; + // 🚨 The attempt stays IN FLIGHT until its outcome is fully weighed (review on #7087). + // Merge allows synchronous re-entry, so a commit published from inside wasUnavailable + // or onStillUnavailable arrives HERE before this branch finishes; marked idle, it + // started a nested attempt before a commit-started failure had disproved commits, and + // that nested Start cleared any retained reconnect β€” an Unavailable answer to it then + // left the write waiting for a reconnect that had already happened. In flight, the + // re-entrant evidence is retained and judged below like any other. var outcome = e.Outcome!; if (!wasUnavailable(outcome)) { state.Done = true; + state.InFlight = false; observer.OnNext(outcome); observer.OnCompleted(); return; @@ -260,6 +272,7 @@ void Start(bool byGap) state.CommitsCount = false; state.PendingCommit = false; } + state.InFlight = false; if (state.PendingGap) Start(byGap: true); else if (state.PendingCommit) diff --git a/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs b/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs index 573b5e2578..ff344719d8 100644 --- a/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs +++ b/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs @@ -304,7 +304,9 @@ private sealed class Rig : IDisposable { private readonly IDisposable _subscription; - public Rig() + /// Runs inside the still-unavailable callback, so a case can + /// make the callback itself produce evidence (it re-enters the sequence synchronously). + public Rig(Action? whenStillUnavailable = null) { Feed = new InProcessMeshChangeFeed(Store); _subscription = Feed @@ -317,10 +319,18 @@ public Rig() return attempt; }, outcome => outcome == Unavailable, - outcome => StillUnavailable = StillUnavailable.Add(outcome)) + outcome => + { + StillUnavailable = StillUnavailable.Add(outcome); + whenStillUnavailable?.Invoke(this); + AttemptsWhenCallbackReturned = AttemptsWhenCallbackReturned.Add(Attempts.Count); + }) .Subscribe(r => Results = Results.Add(r)); } + /// How many attempts existed as each still-unavailable callback returned. + public ImmutableList AttemptsWhenCallbackReturned { get; private set; } = []; + public FaultInjectingStorageAdapter Store { get; } = new(new InMemoryStorageAdapter()); public InProcessMeshChangeFeed Feed { get; } public ImmutableList> Attempts { get; private set; } = []; @@ -418,6 +428,81 @@ public void AVerdict_IsEmitted_AndNothingIsReissuedAfterIt() rig.Attempts.Should().HaveCount(1); } + /// + /// Review on #7087: the still-unavailable callback can itself publish a commit in the partition, + /// which re-enters the sequence synchronously. That commit must not start an attempt while the + /// failed one is still being weighed: a commit-started failure disproves commits, the callback's + /// own commit included, so nothing is re-issued until a reconnect. + /// + [Fact] + public void ACommitFromTheStillUnavailableCallback_IsWeighedAfterTheFailure_NotStartedInsideIt() + { + using var rig = new Rig(r => r.Commit()); + rig.Commit(); + rig.Answer(0, Unavailable); + + rig.AttemptsWhenCallbackReturned.Should().Equal([1], + "no attempt starts from inside the callback that reports the failed one"); + rig.Attempts.Should().HaveCount(1, + "the commit-started failure disproved commits, including the one its own callback made"); + + rig.Gap(); + rig.Attempts.Should().HaveCount(2, "a reconnect still counts"); + rig.Answer(1, Landed); + rig.Results.Should().Equal(Landed); + } + + /// + /// Review on #7087: the same re-entrant commit while a reconnect is RETAINED. The retained + /// reconnect starts the next attempt once the failure is weighed β€” it is not cleared by an attempt + /// the callback's commit started from inside the callback. + /// + [Fact] + public void ACommitFromTheStillUnavailableCallback_KeepsTheRetainedReconnect() + { + using var rig = new Rig(r => r.Commit()); + rig.Commit(); + rig.Gap(); + rig.Answer(0, Unavailable); + + rig.AttemptsWhenCallbackReturned.Should().Equal([1], + "no attempt starts from inside the callback that reports the failed one"); + rig.Attempts.Should().HaveCount(2, "the retained reconnect starts the next attempt"); + rig.Answer(1, Landed); + rig.Results.Should().Equal(Landed); + } + + /// + /// Review on #7087: evidence that arrives WHILE the sequence is being subscribed β€” a feed that + /// announces its reconnect as it is subscribed β€” with an attempt that answers at once. The outcome + /// must reach the sequence; with the evidence connected before the outcomes, it was published into + /// a subject nobody had subscribed yet, and the sequence stayed "in flight" for good. + /// + [Fact] + public void EvidenceDuringSubscription_WithAnImmediateAnswer_IsNotLost() + { + var feed = new GapOnSubscribeFeed(); + var results = ImmutableList.Empty; + using var subscription = feed + .ReissueWhenStoreAnswers("P", () => Observable.Return(Landed), outcome => outcome == Unavailable) + .Subscribe(r => results = results.Add(r)); + + results.Should().Equal([Landed], "the attempt the subscription-time reconnect started answered"); + } + + /// + /// A feed whose backend reconnected as it was subscribed: + /// delivers one declared gap synchronously on subscription. It carries no commits. + /// + private sealed class GapOnSubscribeFeed : IMeshInvalidationFeed + { + public IDisposable Subscribe(Action handler, MeshChangeKind? filter = null) + => System.Reactive.Disposables.Disposable.Empty; + + public IObservable Gaps => Observable.Return(new ChangeFeedGap( + "postgresql:mesh_node_changes", DateTimeOffset.UtcNow.AddSeconds(-1), DateTimeOffset.UtcNow)); + } + [Fact] public void RecordReissueVerdict_ReplacesTheClass_OnlyWhileTheSameObligationStands() { From c83311130ead1b41534224ee01fb46866ce7e03c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= <6334612+rbuergi@users.noreply.github.com> Date: Fri, 23 Oct 2026 21:17:42 +0200 Subject: [PATCH 5/7] fix(release): one owner for an Unavailable stamp; the stamp is evidence; stop on dispose (review on #7087) - CompleteInheritedRelease leaves an inherited Unavailable stamp to the recovery watch: two attempts no longer meet the store at one id, and a verdict is written back rather than only logged. - The watch counts the stamp itself as evidence (a record the store accepted after the refused create): a reconnect or commit that preceded the watch is never replayed to it. A failure of that re-issue disproves nothing. - onStillUnavailable is told whether commits still count, so the warning no longer blames a commit for a reconnect-started attempt. - Disposal stops the state machine first; Start never subscribes a write after. Each new case fails with its fix removed and passes with it. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../LateReleaseAdoption.cs | 27 +++- .../Architecture/ReleaseFailureClasses.md | 12 +- .../StoreReachability.cs | 43 +++++- ...eRecutIsReissuedWhenTheStoreAnswersTest.cs | 145 +++++++++++++++++- 4 files changed, 206 insertions(+), 21 deletions(-) diff --git a/src/MeshWeaver.Compiler.Pipeline/LateReleaseAdoption.cs b/src/MeshWeaver.Compiler.Pipeline/LateReleaseAdoption.cs index 0fb2e0ff4a..a2068921b8 100644 --- a/src/MeshWeaver.Compiler.Pipeline/LateReleaseAdoption.cs +++ b/src/MeshWeaver.Compiler.Pipeline/LateReleaseAdoption.cs @@ -220,6 +220,14 @@ internal static IObservable CompleteInheritedRelease( || InheritedObligation(definition, hubPath, NodeTypeCompilationHelpers.FrameworkVersion) is not { } stamped) return Observable.Empty(); + // 🚨 An inherited UNAVAILABLE stamp has exactly ONE owner on this activation: the + // recovery watch (ReissueWhenTheStoreAnswers), which counts the stamp itself as + // evidence and re-issues at once, and writes a verdict back to the record. Cutting it + // here as well ran two attempts at the same id concurrently, and a verdict this branch + // received was only logged β€” the watch then re-issued a create the store had already + // judged, from the stale class (review on #7087). + if (UnavailableObligation(definition, hubPath, NodeTypeCompilationHelpers.FrameworkVersion) is not null) + return Observable.Empty(); logger?.LogInformation( "[ReleasePostCondition] {HubPath}: this activation inherits a build with no " + "release β€” unreleasedBuildPath={ReleasePath} ({Reason}). Cutting it at that id " @@ -362,7 +370,8 @@ internal static IObservable ReissueWhenTheStoreAnswers( logger?.LogInformation( "[ReleasePostCondition] {HubPath}: the release at {ReleasePath} was not created " + "because the store could not be reached ({Reason}). It is re-issued at that id " - + "as soon as the store is evidenced to answer again β€” a change-feed reconnect or a " + + "now, since the store accepted this stamp after the failed create, and again on " + + "each later piece of evidence that the store answers β€” a change-feed reconnect or a " + "commit in partition '{Partition}' (#7072).", hubPath, stamped, definition.UnreleasedBuildReason ?? "(no reason recorded)", partition); var pending = x.Node with { Content = definition with { RequestedReleaseBy = null } }; @@ -374,12 +383,18 @@ internal static IObservable ReissueWhenTheStoreAnswers( hub, hubPath, build, pending, definition.LastCompilationActivityPath, logger, reusePath: stamped)), outcome => outcome.WasUnavailable, - outcome => logger?.LogWarning( + (outcome, commitsStillCount) => logger?.LogWarning( "[ReleasePostCondition] {HubPath}: re-issuing the release at {ReleasePath} " - + "was answered Unavailable again{Because}. A commit landed in the partition " - + "and still the store never saw this create, so the next re-issue waits for " - + "the change feed to reconnect (#7072).", - hubPath, stamped, outcome.Because)) + + "was answered Unavailable again{Because}. The next re-issue waits for {Next} (#7072).", + hubPath, stamped, outcome.Because, + commitsStillCount + ? $"a change-feed reconnect or a commit in partition '{partition}'" + : "a change-feed reconnect: this re-issue was started by a commit in the " + + "partition and the store still never saw this create"), + // The stamp is itself a record the store accepted after the create failed, and + // a reconnect or commit that preceded it is never replayed to this watch β€” an + // idle partition would leave the release unpublished (review on #7087). + storeAnsweredBeforeSubscription: true) .SelectMany(outcome => { if (outcome.ReleasePath is not { } landed) diff --git a/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md b/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md index a58e5303ae..b3636fcc29 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md @@ -590,7 +590,17 @@ The log line said the retry was meaningful, and no component on the live pod mad subscribed cannot start an attempt whose answer has nowhere to go. And a failed attempt stays in flight until its outcome has been weighed, so a commit published from inside the still-unavailable callback is retained and judged like any other evidence. It never starts a - nested attempt that would clear a retained reconnect. + nested attempt that would clear a retained reconnect. The stamp itself is evidence too: it is a + record the store accepted after the refused create, and a reconnect or commit that came before + the watch saw the stamp is never replayed to it. So the watch makes its first re-issue as soon as + it sees the stamp. That re-issue proves nothing if it is answered `Unavailable` (the stamp may + have been accepted only locally), so commits keep counting after it. The warning for a re-issue + answered `Unavailable` names the evidence that will start the next one. It names a commit as + the cause only when a commit actually started that re-issue. An activation that INHERITS an + `Unavailable` stamp leaves it to this watch alone: the inherited cut (#6056) skips that class. + Otherwise two attempts met the store at the same id, and a verdict the inherited cut received + was only logged, so the watch would later re-issue a create the store had already judged. A + watch that is disposed never starts another write, even from evidence it had retained. 4. **A verdict on a re-issue is written back.** If the store comes back and answers the re-issue with a verdict (`Refused`, `Faulted`), `RecordReissueVerdict` replaces the stamp's reason and class with that verdict's, through the owner stream, and only while the record still holds the diff --git a/src/MeshWeaver.Mesh.Contract/StoreReachability.cs b/src/MeshWeaver.Mesh.Contract/StoreReachability.cs index 2ef8f3b824..7ce2e9e2f6 100644 --- a/src/MeshWeaver.Mesh.Contract/StoreReachability.cs +++ b/src/MeshWeaver.Mesh.Contract/StoreReachability.cs @@ -169,20 +169,33 @@ private static IObservable EvidenceBySource(IMeshInvalidationFeed? feed, s /// rejects is emitted and the sequence completes, so a refusal /// stays a named failure. It never runs two attempts at once. And it never issues the FIRST /// attempt β€” the caller made it, and is here because it came back Unavailable. + /// + /// Evidence the caller already holds. A caller that learns of the obligation only + /// from a record the store accepted AFTER the write failed (a stamp written after the refused + /// create) holds evidence the feed will never replay: a reconnect or a commit that preceded the + /// subscription is gone. It passes , and the + /// first re-issue starts at once (review on #7087). That evidence disproves nothing if the + /// re-issue is answered Unavailable β€” the record may have been accepted locally ahead of the + /// store β€” so commits keep counting, exactly as after a reconnect-started failure. /// /// The write's outcome. /// The mesh's invalidation feed, or null (then nothing is re-issued). /// The partition the write lands in. /// The SAME write, cold β€” subscribed once per re-issue. /// True when an outcome means "not attempted, retry meaningful". - /// Told about each re-issue that was answered Unavailable again. + /// Told about each re-issue that was answered Unavailable again, + /// with whether a commit in the partition still starts the next one (false: only a + /// reconnect does, because this failure was started by a commit). + /// The caller holds evidence, from before this + /// subscription, that the store answered after the write failed: start the first re-issue at once. /// The first outcome that is not Unavailable; never completes while the store stays away. public static IObservable ReissueWhenStoreAnswers( this IMeshInvalidationFeed? feed, string partition, Func> attempt, Func wasUnavailable, - Action? onStillUnavailable = null) + Action? onStillUnavailable = null, + bool storeAnsweredBeforeSubscription = false) { if (feed is null) return Observable.Never(); @@ -198,6 +211,12 @@ public static IObservable ReissueWhenStoreAnswers( void Start(bool byGap) { + // 🚨 Never start a write once the subscription is disposed (review on #7087). A + // callback may dispose it and then return into this path with evidence retained; + // assigning a slot to a disposed SerialDisposable disposes the slot, but only AFTER + // the cold write below would already have been subscribed. + if (state.Stopped) + return; state.InFlight = true; state.PendingGap = false; state.PendingCommit = false; @@ -226,13 +245,17 @@ void Start(bool byGap) // first, that evidence ran Start before the hot `outcomes` subject had a subscriber, an // attempt that answered in that window lost its outcome, and InFlight stayed true for good // β€” every later piece of evidence was retained and none ever started an attempt. + // Evidence the caller holds from before the subscription is delivered first, after the + // outcomes are connected, as the non-disproving kind (see the summary). + var evidence = EvidenceBySource(feed, partition); + if (storeAnsweredBeforeSubscription) + evidence = evidence.StartWith(true); var events = outcomes.Select(x => (IsOutcome: true, x.ByGap, Outcome: (T?)x.Outcome)) - .Merge(EvidenceBySource(feed, partition) - .Select(byGap => (IsOutcome: false, ByGap: byGap, Outcome: default(T)))); + .Merge(evidence.Select(byGap => (IsOutcome: false, ByGap: byGap, Outcome: default(T)))); var subscription = events.Subscribe(e => { - if (state.Done) + if (state.Done || state.Stopped) return; if (!e.IsOutcome) { @@ -265,7 +288,7 @@ void Start(bool byGap) observer.OnCompleted(); return; } - onStillUnavailable?.Invoke(outcome); + onStillUnavailable?.Invoke(outcome, state.CommitsCount && e.ByGap); if (!e.ByGap) { // A commit started it and the store still never saw it: commits are disproved. @@ -279,7 +302,10 @@ void Start(bool byGap) Start(byGap: false); }, observer.OnError); - return new CompositeDisposable(subscription, inFlight, outcomes); + // The stop flag goes FIRST, so the state machine is stopped before anything it owns is + // torn down β€” including a disposal that happens from inside one of its own callbacks. + return new CompositeDisposable( + Disposable.Create(() => state.Stopped = true), subscription, inFlight, outcomes); }); } @@ -292,6 +318,9 @@ private sealed class ReissueState public bool PendingGap; public bool PendingCommit; public bool Done; + + /// Set when the subscription is disposed; read on every transition, from any thread. + public volatile bool Stopped; } private static bool InPartition(string? path, string partition) diff --git a/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs b/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs index ff344719d8..e018aa2cd3 100644 --- a/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs +++ b/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs @@ -110,7 +110,8 @@ await MeshService.CreateNode(MeshNode.FromPath(typePath) with /// settle's terminal stamp, through the production field-set (ApplyCompileSuccess). /// private async Task RecutRefusedAndStampAsync( - string typePath, NodeCompilationResult build, string stamped, MeshNode seeded, string? classOverride = null) + string typePath, NodeCompilationResult build, string stamped, MeshNode seeded, string? classOverride = null, + Action? beforeStamp = null) { var ct = TestContext.Current.CancellationToken; var logger = Mesh.ServiceProvider.GetRequiredService().CreateLogger("MeshWeaver.Graph.CompileWatcher"); @@ -129,6 +130,7 @@ await MeshService.CreateNode(MeshNode.FromPath(typePath) with "a create the store never saw is classified Unavailable AT THE ARM, from the typed rejection " + "reason β€” never parsed out of the sentence"); + beforeStamp?.Invoke(); await Mesh.GetMeshNodeStream(typePath).Update(current => { var def = current.ContentAs(Mesh.JsonSerializerOptions)!; @@ -212,6 +214,66 @@ await Mesh.GetMeshNodeStream(typePath) .Should().Be(736L, "the release names the recorded build"); } + /// + /// Review on #7087: an activation that INHERITS an Unavailable stamp has one owner for it. The + /// store has come back and judges the create: exactly one attempt reaches it, the verdict is + /// written to the record, and later recovery evidence re-issues nothing. With the inherited cut + /// (#6056) also running, two attempts met the store and the inherited one's verdict was only + /// logged. + /// + [Fact(Timeout = 120_000)] + public async Task AnInheritedUnavailableStamp_HasOneOwner_AndItsVerdictIsRecorded() + { + var ct = TestContext.Current.CancellationToken; + var typePath = $"{TestPartition}/PrInherit{Guid.NewGuid().ToString("N")[..8]}"; + var bytes = File.ReadAllBytes(typeof(AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest).Assembly.Location); + var location = await Mesh.ServiceProvider.GetRequiredService() + .PutWithLocation(typePath, 736, bytes, null) + .Should().Within(TestTimeouts.WriteConvergence).Emit(cancellationToken: ct); + var definition = new NodeTypeDefinition + { + Configuration = "config => config", + CompilationStatus = CompilationStatus.Ok, + LastCompiledVersion = 736, + LatestReleasePath = $"{typePath}/{PreviousRelease}", + LatestAssemblyCollection = location.Collection, + LatestAssemblyPath = location.ContentPath, + CompiledFrameworkVersion = NodeTypeCompilationHelpers.FrameworkVersion, + CompiledSources = ImmutableDictionary.Empty, + }; + var build = LateReleaseAdoption.RecordedBuild(definition)!; + var stamped = $"{typePath}/{GraphNodeTypeNames.ReleaseSegment}/20261022212254-{NodeTypeBuildState.ContentHashOf(build)}"; + + using var judged = Storage.RefuseWritesAsJudged(stamped); + await MeshService.CreateNode(MeshNode.FromPath(typePath) with + { + Name = typePath, + NodeType = MeshNode.NodeTypePath, + State = MeshNodeState.Active, + Content = definition with + { + UnreleasedBuildPath = stamped, + UnreleasedBuildReason = "the data store was unreachable (an earlier activation)", + UnreleasedBuildFailureClass = ReleaseFailureClass.Unavailable, + }, + }) + .Should().Within(TestTimeouts.Convergence) + .Emit("the NodeType exists with an inherited Unavailable stamp", cancellationToken: ct); + + var recorded = await Mesh.GetMeshNodeStream(typePath) + .Should().Within(TestTimeouts.CrossSilo) + .Match(n => n.ContentAs(Mesh.JsonSerializerOptions) is { } d + && d.UnreleasedBuildFailureClass == ReleaseFailureClass.Refused, + "the activation's one attempt met the verdict and wrote it to the record", ct); + recorded.ContentAs(Mesh.JsonSerializerOptions)!.UnreleasedBuildPath + .Should().Be(stamped, "this build still has no release"); + + TheStoreAnswersAgain(); + await judged.Arrivals.Skip(1) + .Should().NotEmit(TestTimeouts.Quick, + "one owner, one attempt: no second cut at the same id, and no re-issue of a judged create"); + } + /// /// Negative control 2: a stamp the store DID judge (class ) /// is a named failure, never re-issued β€” the same refusal and the same recovery signals as the @@ -257,13 +319,17 @@ public async Task AReissueTheStoreRefuses_RecordsTheRefusal_AndIsNotReissuedAgai var typePath = $"{TestPartition}/TriageStatus{Guid.NewGuid().ToString("N")[..8]}"; var (build, stamped, seeded) = await SeedActivatedAsync(typePath); + // The store comes back JUDGING between the refused create and the stamp: the stamp is itself + // evidence the store answers, so the watch re-issues as soon as it sees it, and that re-issue + // meets the verdict (a judged write takes precedence over the unreachable switch). + FaultSwitch? judgedSwitch = null; using (Storage.RefuseWritesAsUnreachable(stamped)) - await RecutRefusedAndStampAsync(typePath, build, stamped, seeded); + await RecutRefusedAndStampAsync(typePath, build, stamped, seeded, + beforeStamp: () => judgedSwitch = Storage.RefuseWritesAsJudged(stamped)); - var judged = Storage.RefuseWritesAsJudged(stamped); + var judged = judgedSwitch!; try { - TheStoreAnswersAgain(); await judged.Arrivals.Should().Within(TestTimeouts.Convergence) .Emit("the store came back and the re-issue reached it", ct); @@ -306,7 +372,7 @@ private sealed class Rig : IDisposable /// Runs inside the still-unavailable callback, so a case can /// make the callback itself produce evidence (it re-enters the sequence synchronously). - public Rig(Action? whenStillUnavailable = null) + public Rig(Action? whenStillUnavailable = null, bool storeAnsweredBeforeSubscription = false) { Feed = new InProcessMeshChangeFeed(Store); _subscription = Feed @@ -319,15 +385,20 @@ public Rig(Action? whenStillUnavailable = null) return attempt; }, outcome => outcome == Unavailable, - outcome => + (outcome, commitsStillCount) => { StillUnavailable = StillUnavailable.Add(outcome); + CommitsStillCount = CommitsStillCount.Add(commitsStillCount); whenStillUnavailable?.Invoke(this); AttemptsWhenCallbackReturned = AttemptsWhenCallbackReturned.Add(Attempts.Count); - }) + }, + storeAnsweredBeforeSubscription) .Subscribe(r => Results = Results.Add(r)); } + /// What each still-unavailable callback was told: does a commit still start a re-issue. + public ImmutableList CommitsStillCount { get; private set; } = []; + /// How many attempts existed as each still-unavailable callback returned. public ImmutableList AttemptsWhenCallbackReturned { get; private set; } = []; @@ -345,6 +416,9 @@ public void Gap() => Store.DeliverChangeFeedGap(new ChangeFeedGap( public void Answer(int attempt, string outcome) => Attempts[attempt].OnNext(outcome); + /// Disposes the re-issue subscription, as its owner going away does. + public void StopWatching() => _subscription.Dispose(); + public void Dispose() { _subscription.Dispose(); @@ -415,6 +489,63 @@ public void ACommitStartedFailure_LeavesOnlyReconnectsAsEvidence() rig.Results.Should().Equal(Landed); } + /// + /// Review on #7087: the callback is told which evidence still counts. A reconnect-started failure + /// leaves commits counting; a commit-started one leaves only reconnects β€” so a caller's report + /// never claims a commit started an attempt that a reconnect started. + /// + [Fact] + public void TheStillUnavailableCallback_IsToldWhetherCommitsStillCount() + { + using var rig = new Rig(); + rig.Gap(); + rig.Answer(0, Unavailable); + rig.Commit(); + rig.Answer(1, Unavailable); + rig.CommitsStillCount.Should().Equal([true, false], + "a reconnect-started failure disproves nothing; a commit-started one disproves commits"); + } + + /// + /// Review on #7087: a caller that already holds evidence from before the subscription (the stamp, + /// accepted by the store after the refused create) starts the first re-issue at once β€” the feed + /// never replays what preceded it. That evidence disproves nothing if the re-issue fails. + /// + [Fact] + public void EvidenceHeldBeforeSubscription_StartsTheFirstReissue_AndDisprovesNothing() + { + using var rig = new Rig(storeAnsweredBeforeSubscription: true); + rig.Attempts.Should().HaveCount(1, "the evidence the caller holds starts a re-issue with no feed event"); + rig.Answer(0, Unavailable); + rig.CommitsStillCount.Should().Equal([true], "evidence held before the subscription disproves nothing"); + rig.Commit(); + rig.Attempts.Should().HaveCount(2, "commits still count"); + rig.Answer(1, Landed); + rig.Results.Should().Equal(Landed); + } + + /// + /// Review on #7087: a subscription disposed from inside the still-unavailable callback, with a + /// reconnect retained, starts no further write β€” the attempt factory is not invoked again. + /// + [Fact] + public void DisposedFromTheCallback_WithAGapPending_StartsNoFurtherWrite() + { + using var rig = new Rig(r => r.StopWatching()); + rig.Commit(); + rig.Gap(); + rig.Answer(0, Unavailable); + rig.Attempts.Should().HaveCount(1, "a disposed re-issue never subscribes another write"); + } + + /// Without that evidence, nothing is re-issued before the feed says the store answers. + [Fact] + public void WithoutEvidence_NothingIsReissued() + { + using var rig = new Rig(); + rig.Attempts.Should().BeEmpty("the caller made the first attempt; only evidence starts a re-issue"); + } + /// A verdict ends the sequence: it is emitted, and no evidence after it starts anything. [Fact] public void AVerdict_IsEmitted_AndNothingIsReissuedAfterIt() From 9cc37421b475438bfeb0f60ee8ac7f5d1fd0e9db Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= <6334612+rbuergi@users.noreply.github.com> Date: Sat, 24 Oct 2026 08:11:48 +0200 Subject: [PATCH 6/7] fix(release): deliver held re-issue evidence after the live feed is connected (review on #7087) StartWith(true) emitted the caller's held evidence before EvidenceBySource was subscribed, so a reconnect delivered while that first attempt was being subscribed was lost to the non-replaying feed; an Unavailable answer then left an idle partition with nothing to start the next attempt. The held evidence is now merged in after the live commits and gaps are connected, through the same serialised pipeline. AGapDuringTheHeldEvidenceAttempt_IsRetained_AndStartsTheNextAttempt fails without the fix (1 attempt, expected 2) and passes with it. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../StoreReachability.cs | 10 ++++--- ...eRecutIsReissuedWhenTheStoreAnswersTest.cs | 26 ++++++++++++++++++- 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/src/MeshWeaver.Mesh.Contract/StoreReachability.cs b/src/MeshWeaver.Mesh.Contract/StoreReachability.cs index 7ce2e9e2f6..4acdfcef67 100644 --- a/src/MeshWeaver.Mesh.Contract/StoreReachability.cs +++ b/src/MeshWeaver.Mesh.Contract/StoreReachability.cs @@ -245,11 +245,15 @@ void Start(bool byGap) // first, that evidence ran Start before the hot `outcomes` subject had a subscriber, an // attempt that answered in that window lost its outcome, and InFlight stayed true for good // β€” every later piece of evidence was retained and none ever started an attempt. - // Evidence the caller holds from before the subscription is delivered first, after the - // outcomes are connected, as the non-disproving kind (see the summary). + // Evidence the caller holds from before the subscription is delivered LAST, as the + // non-disproving kind (see the summary): after the outcomes AND the live feed are + // connected, through the same serialised Merge (review on #7087). StartWith emitted it + // before the live feed was subscribed, so a reconnect that arrived while that first + // attempt was being subscribed was lost to the non-replaying feed, and an Unavailable + // answer then left an idle partition with nothing to start the next attempt. var evidence = EvidenceBySource(feed, partition); if (storeAnsweredBeforeSubscription) - evidence = evidence.StartWith(true); + evidence = evidence.Merge(Observable.Return(true)); var events = outcomes.Select(x => (IsOutcome: true, x.ByGap, Outcome: (T?)x.Outcome)) .Merge(evidence.Select(byGap => (IsOutcome: false, ByGap: byGap, Outcome: default(T)))); diff --git a/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs b/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs index e018aa2cd3..37b8567e03 100644 --- a/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs +++ b/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs @@ -372,7 +372,10 @@ private sealed class Rig : IDisposable /// Runs inside the still-unavailable callback, so a case can /// make the callback itself produce evidence (it re-enters the sequence synchronously). - public Rig(Action? whenStillUnavailable = null, bool storeAnsweredBeforeSubscription = false) + /// Runs as each attempt is subscribed, with its index, so a case + /// can make the feed deliver evidence DURING that subscription. + public Rig(Action? whenStillUnavailable = null, bool storeAnsweredBeforeSubscription = false, + Action? whenAttemptStarts = null) { Feed = new InProcessMeshChangeFeed(Store); _subscription = Feed @@ -382,6 +385,7 @@ public Rig(Action? whenStillUnavailable = null, bool storeAnsweredBeforeSub { var attempt = new System.Reactive.Subjects.Subject(); Attempts = Attempts.Add(attempt); + whenAttemptStarts?.Invoke(this, Attempts.Count - 1); return attempt; }, outcome => outcome == Unavailable, @@ -538,6 +542,26 @@ public void DisposedFromTheCallback_WithAGapPending_StartsNoFurtherWrite() rig.Attempts.Should().HaveCount(1, "a disposed re-issue never subscribes another write"); } + /// + /// Review on #7087: the live feed is connected BEFORE the evidence the caller already holds starts + /// the first attempt. A reconnect delivered while that attempt is being subscribed is retained, + /// and when the attempt answers Unavailable it starts the next one with no further feed event β€” on + /// an idle partition nothing else ever would. + /// + [Fact] + public void AGapDuringTheHeldEvidenceAttempt_IsRetained_AndStartsTheNextAttempt() + { + using var rig = new Rig(storeAnsweredBeforeSubscription: true, + whenAttemptStarts: (r, index) => { if (index == 0) r.Gap(); }); + rig.Attempts.Should().HaveCount(1, "the held evidence starts the first re-issue"); + + rig.Answer(0, Unavailable); + rig.Attempts.Should().HaveCount(2, + "the reconnect that arrived while the first attempt was subscribing is retained, not lost"); + rig.Answer(1, Landed); + rig.Results.Should().Equal(Landed); + } + /// Without that evidence, nothing is re-issued before the feed says the store answers. [Fact] public void WithoutEvidence_NothingIsReissued() From 30f4e666b3fa45d4f2262f273c929dd7c22f249e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= <6334612+rbuergi@users.noreply.github.com> Date: Sat, 24 Oct 2026 08:31:43 +0200 Subject: [PATCH 7/7] fix(release): only commits the store announced count as re-issue evidence (review on #7087) The invalidation feed also carries optimistic notifications: a compile settle publishes its own-stream update before the persistence sampler stores it. One of those starting a re-issue that is answered Unavailable disproved commits for good, so the durable commit that follows when write connections recover could never start the next attempt. IMeshInvalidationFeed gains DurableCommits (default: never), fed by InProcessMeshChangeFeed only from the storage relay path, and the evidence reads it instead of every invalidation. AnOptimisticNotification_IsNotEvidence_AndADurableCommitStillStartsTheReissue fails without the fix and passes with it. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Architecture/ReleaseFailureClasses.md | 6 +++- src/MeshWeaver.Hosting/MeshChangeFeed.cs | 33 +++++++++++++++++-- .../Services/IMeshChangeFeed.cs | 13 ++++++++ .../StoreReachability.cs | 14 +++++--- ...eRecutIsReissuedWhenTheStoreAnswersTest.cs | 26 ++++++++++++++- 5 files changed, 83 insertions(+), 9 deletions(-) diff --git a/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md b/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md index b3636fcc29..3aa01fd17e 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/ReleaseFailureClasses.md @@ -576,7 +576,11 @@ The log line said the retry was meaningful, and no component on the live pod mad 3. **It waits for evidence, never for time.** `StoreReachability.ReissueWhenStoreAnswers` in `MeshWeaver.Mesh.Contract` waits for `StoreAnswersAgain`. That signal fires on a declared `ChangeFeedGap`, which is the LISTEN connection resuming after a database restart, or on a commit - announced in the type's partition, which the store must have accepted. While the store is away, + that the STORE announced in the type's partition (`IMeshInvalidationFeed.DurableCommits`), which + it must have accepted. A notification published directly on the feed is not counted, because it + may precede persistence: a compile settle publishes its own-stream update before the persistence + sampler stores it. Counting it could start a re-issue that disproves commits before any durable + commit arrives. While the store is away, neither fires and nothing is re-issued. Evidence is observed for the whole life of the watch: evidence that arrives while an attempt is in flight is kept (the event that started the attempt is not), and starts the next attempt as soon as that one is answered `Unavailable`. A feed does diff --git a/src/MeshWeaver.Hosting/MeshChangeFeed.cs b/src/MeshWeaver.Hosting/MeshChangeFeed.cs index 923b0f4ae4..03e5519d44 100644 --- a/src/MeshWeaver.Hosting/MeshChangeFeed.cs +++ b/src/MeshWeaver.Hosting/MeshChangeFeed.cs @@ -23,6 +23,8 @@ public class InProcessMeshChangeFeed : IMeshChangeFeed, IMeshInvalidationFeed, I private readonly ISubject _subject; private readonly ISubject _invalidations; private readonly StorageChangeFeedRelay? _storageRelay; + private readonly Subject _durableLifetime = new(); + private readonly ISubject _durable; private readonly Subject _gapsLifetime = new(); private readonly ISubject _gaps; private readonly IDisposable? _gapRelay; @@ -40,6 +42,7 @@ public InProcessMeshChangeFeed() _subject = Subject.Synchronize(_subjectLifetime); _invalidations = Subject.Synchronize(_invalidationLifetime); _gaps = Subject.Synchronize(_gapsLifetime); + _durable = Subject.Synchronize(_durableLifetime); } /// @@ -139,10 +142,34 @@ public void PublishLocal(MeshChangeEvent change) /// internal void PublishInvalidation(MeshChangeEvent change) { - if (!_disposed) - _invalidations.OnNext(change); + if (_disposed) + return; + _invalidations.OnNext(change); + // Only this path is a commit the store announced; a direct Publish is not (DurableCommits). + _durable.OnNext(change); } + /// + /// Each subscriber isolated, as is: one that throws must not keep the + /// next from seeing the commit. + public IObservable DurableCommits => Observable.Create(observer => + _durable.Subscribe( + change => + { + try + { + observer.OnNext(change); + } + catch (Exception ex) + { + logger?.LogError(ex, + "Mesh durable-commit subscriber failed for {Kind} {Path}; continuing with the others", + change.Kind, change.Path); + } + }, + observer.OnError, + observer.OnCompleted)); + /// /// Subscribes a handler to mesh change events, optionally filtered by change kind. /// @@ -199,6 +226,8 @@ public void Dispose() _gapRelay?.Dispose(); _gaps.OnCompleted(); _gapsLifetime.Dispose(); + _durable.OnCompleted(); + _durableLifetime.Dispose(); _subject.OnCompleted(); _invalidations.OnCompleted(); _subjectLifetime.Dispose(); diff --git a/src/MeshWeaver.Mesh.Contract/Services/IMeshChangeFeed.cs b/src/MeshWeaver.Mesh.Contract/Services/IMeshChangeFeed.cs index fef8ce116a..fa0ea5be85 100644 --- a/src/MeshWeaver.Mesh.Contract/Services/IMeshChangeFeed.cs +++ b/src/MeshWeaver.Mesh.Contract/Services/IMeshChangeFeed.cs @@ -53,6 +53,19 @@ public interface IMeshInvalidationFeed /// Default: never emits β€” a feed with no cross-process source cannot have a gap. /// IObservable Gaps => System.Reactive.Linq.Observable.Never(); + + /// + /// Only the commits the STORAGE BACKEND announced β€” each one already durably stored, whichever + /// process made it. A direct is NOT here: it reaches + /// as an optimistic invalidation and may precede persistence (a compile + /// settle publishes its own-stream update before the persistence sampler stores it), so it is + /// no evidence that the store accepted anything (review on #7087). + /// + /// Default: never emits β€” a feed that cannot tell a durable commit from an optimistic + /// notification offers no durable evidence at all, rather than passing the one off as the + /// other. + /// + IObservable DurableCommits => System.Reactive.Linq.Observable.Never(); } /// diff --git a/src/MeshWeaver.Mesh.Contract/StoreReachability.cs b/src/MeshWeaver.Mesh.Contract/StoreReachability.cs index 4acdfcef67..9bcc3eaedb 100644 --- a/src/MeshWeaver.Mesh.Contract/StoreReachability.cs +++ b/src/MeshWeaver.Mesh.Contract/StoreReachability.cs @@ -111,8 +111,10 @@ public static string DescribeMayHavePartiallyLanded(string operation) /// /// The EVIDENCE that the store answers again, as this process can observe it (#7072): each /// declared β€” the backend's cross-process feed lost its connection - /// and RESUMED, which a database restart always produces β€” and each commit announced in - /// , since a commit is announced only after the store accepted it. + /// and RESUMED, which a database restart always produces β€” and each commit THE STORE announced + /// in (), since + /// such a commit is announced only after the store accepted it. An optimistic notification + /// published before persistence is not evidence and is not counted. /// /// 🚨 Facts, never a clock. Nothing here fires because time passed; while the store is /// unreachable no commit lands and no feed resumes, so the signal stays silent exactly as long as @@ -136,9 +138,11 @@ private static IObservable EvidenceBySource(IMeshInvalidationFeed? feed, s { if (feed is null) return Observable.Never(); - // The handler is forwarded as-is and the filter runs as an operator, so a fault in it reaches - // the subscriber's OnError instead of being thrown out of the feed's delivery callback. - var commits = Observable.Create(observer => feed.Subscribe(observer.OnNext)) + // 🚨 DURABLE commits only (review on #7087). The invalidation feed also carries optimistic + // notifications published before persistence; one of those starting a re-issue that is + // answered Unavailable would disprove commits for good, and the durable commit that follows + // when the store's write connections recover could then never start the next one. + var commits = feed.DurableCommits .Where(change => partition is null || InPartition(change.Path, partition)) .Select(_ => false); return commits.Merge(feed.Gaps.Select(_ => true)); diff --git a/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs b/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs index 37b8567e03..5d0117d4c0 100644 --- a/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs +++ b/test/MeshWeaver.Compiler.Pipeline.Test/AnUnavailableReleaseRecutIsReissuedWhenTheStoreAnswersTest.cs @@ -412,9 +412,15 @@ public Rig(Action? whenStillUnavailable = null, bool storeAnsweredBeforeSub public ImmutableList Results { get; private set; } = []; public ImmutableList StillUnavailable { get; private set; } = []; - public void Commit() => Feed.Publish(new MeshChangeEvent( + /// A commit the STORE announced in the partition (the durable relay's path). + public void Commit() => Feed.PublishLocal(new MeshChangeEvent( "P", "other", "P/other", MeshChangeKind.Updated, null, 1, DateTimeOffset.UtcNow)); + /// An OPTIMISTIC notification in the partition: a direct publish that may precede + /// persistence, as a compile settle's own-stream update is. + public void OptimisticCommit() => Feed.Publish(new MeshChangeEvent( + "P", "stamp", "P/stamp", MeshChangeKind.Updated, null, 1, DateTimeOffset.UtcNow)); + public void Gap() => Store.DeliverChangeFeedGap(new ChangeFeedGap( "postgresql:mesh_node_changes", DateTimeOffset.UtcNow.AddSeconds(-1), DateTimeOffset.UtcNow)); @@ -562,6 +568,24 @@ public void AGapDuringTheHeldEvidenceAttempt_IsRetained_AndStartsTheNextAttempt( rig.Results.Should().Equal(Landed); } + /// + /// Review on #7087: an optimistic notification β€” a direct publish that may precede persistence β€” + /// is no evidence the store answers. It starts nothing, so it cannot disprove commits, and the + /// durable commit that follows starts the re-issue with no gap in between. + /// + [Fact] + public void AnOptimisticNotification_IsNotEvidence_AndADurableCommitStillStartsTheReissue() + { + using var rig = new Rig(); + rig.OptimisticCommit(); + rig.Attempts.Should().BeEmpty("a notification published before persistence proves nothing about the store"); + + rig.Commit(); + rig.Attempts.Should().HaveCount(1, "a commit the store announced is evidence"); + rig.Answer(0, Landed); + rig.Results.Should().Equal(Landed); + } + /// Without that evidence, nothing is re-issued before the feed says the store answers. [Fact] public void WithoutEvidence_NothingIsReissued()