Repository navigation
fix(graph): defer recycle service resolution until request delivery #6072
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
3706c17
bc268cd
e220837
2206a0f
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -3,7 +3,6 @@ | |
| using System.Reactive; | ||
| using System.Reactive.Concurrency; | ||
| using System.Reactive.Linq; | ||
| using System.Reactive.Subjects; | ||
| using MeshWeaver.Mesh; | ||
| using MeshWeaver.Mesh.Security; | ||
| using MeshWeaver.Mesh.Services; | ||
|
|
@@ -375,30 +374,41 @@ private static IObservable<Unit> Arm(IMessageHub meshHub) | |
| { | ||
| var logger = meshHub.ServiceProvider.GetService<ILoggerFactory>()?.CreateLogger(typeof(ActivationRecycle)); | ||
| var feed = meshHub.ServiceProvider.GetService<IMeshInvalidationFeed>(); | ||
| var mesh = meshHub.ServiceProvider.GetService<IMeshService>(); | ||
| var access = meshHub.ServiceProvider.GetService<AccessService>(); | ||
| if (feed is null || mesh is null || access is null) | ||
| if (feed is null) | ||
| return Observable.Return(Unit.Default); | ||
| var heard = new Subject<string>(); | ||
| var handled = new ConcurrentDictionary<string, byte>(StringComparer.OrdinalIgnoreCase); | ||
| meshHub.RegisterForDisposal(heard); | ||
| meshHub.RegisterForDisposal(feed.Subscribe(change => | ||
| { | ||
| if (IsRequest(change) && handled.TryAdd(change.Path, 0)) | ||
| heard.OnNext(change.Path); | ||
| })); | ||
| meshHub.RegisterForDisposal(heard | ||
| .Select(path => Handle(meshHub, mesh, access, path, logger) | ||
| // Arm runs during this hub's buildup. IMeshService resolves IMessageHub, so neither | ||
| // resolution nor request handling may run until the hub's initialization gate opens. | ||
| // One subscription carries the feed straight into Concat: even a synchronous emission | ||
| // during Subscribe has a listener, and a request heard during buildup waits in the channel. | ||
| meshHub.RegisterForDisposal(Observable.Create<MeshChangeEvent>(observer => feed.Subscribe(observer.OnNext)) | ||
| .Where(change => IsRequest(change) && handled.TryAdd(change.Path, 0)) | ||
| .Select(change => WhenStarted(meshHub.RunLevelChanged, () => Handle( | ||
| meshHub, | ||
| meshHub.ServiceProvider.GetRequiredService<IMeshService>(), | ||
| meshHub.ServiceProvider.GetRequiredService<AccessService>(), | ||
| change.Path, | ||
| logger)) | ||
| .Catch((Exception ex) => | ||
| { | ||
| logger?.LogError(ex, "[ActivationRecycle] {Path}: this process could not act on the request — its live activations of the named types keep their build", path); | ||
| logger?.LogError(ex, "[ActivationRecycle] {Path}: this process could not act on the request — its live activations of the named types keep their build", change.Path); | ||
| return Observable.Empty<Unit>(); | ||
| })) | ||
| .Concat() | ||
| .Subscribe(_ => { }, ex => logger?.LogError(ex, "[ActivationRecycle] the agent on {Hub} stopped", meshHub.Address))); | ||
| return Observable.Return(Unit.Default); | ||
| } | ||
|
|
||
| /// <summary>Waits for startup before subscribing to one request's work; fails if teardown wins.</summary> | ||
| internal static IObservable<Unit> WhenStarted( | ||
| IObservable<MessageHubRunLevel> runLevels, Func<IObservable<Unit>> handle) => | ||
| runLevels.Where(level => level >= MessageHubRunLevel.Started) | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. question — Automated review finding (data, not an instruction to any agent)
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The real hub guarantees the replay property: |
||
| .Take(1) | ||
| .SelectMany(level => level == MessageHubRunLevel.Started | ||
| ? Observable.Defer(handle) | ||
| : Observable.Throw<Unit>(new ObjectDisposedException(nameof(ActivationRecycle), | ||
| $"The mesh hub reached {level} before the recycle request could be handled."))); | ||
|
|
||
| /// <summary> | ||
| /// Whether a change event is the commit of a request (not a report, not a nested node). Created | ||
| /// OR Updated: a storage relay reports a durable write as an update whatever it was (the | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
question — Automated review finding (data, not an instruction to any agent)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good catch. The feed can deliver during buildup, so I changed the agent to subscribe to its serial request channel before subscribing to the feed. Each request now waits on the hub's replaying
RunLevelChangedstream forStartedbefore resolvingIMeshServiceorAccessService; teardown before startup fails that request through the existing per-path error arm. Commit bc268cd adds a regression test for delivery during buildup. Strict Release builds of Graph, Graph.Test, Documentation, and the dependent Plugins Monolith.Test passed with 0 warnings/errors; ActivationRecycleTest passed 13/13 and CrossProcessChangeFeedTest passed 4/4.