Skip to content

Commit 3e95af7

Browse files
louis4lilouis4li
authored
Merge pull request #1830 from aevatarAI/fix/2026-06-05_stream-topology-storage-conflict
Fix stream topology storage conflicts
2 parents e71546f + 5dfd122 commit 3e95af7

18 files changed

Lines changed: 528 additions & 43 deletions

‎src/Aevatar.CQRS.Projection.Core/DependencyInjection/EventSinkProjectionRuntimeRegistration.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ public static IServiceCollection AddEventSinkProjectionRuntimeCore<TContext, TRu
6464
sp.GetService<Aevatar.Foundation.Abstractions.TypeSystem.IAgentTypeVerifier>(),
6565
sp.GetService<IStreamPubSubMaintenance>(),
6666
sp.GetService<ILoggerFactory>(),
67-
streams: sp.GetService<IStreamProvider>()));
67+
sp.GetService<IStreamForwardingRegistry>()));
6868
services.TryAddSingleton<IProjectionScopeReleaseService<TRuntimeLease>>(sp =>
6969
new ProjectionScopeReleaseService<
7070
TRuntimeLease,

‎src/Aevatar.CQRS.Projection.Core/DependencyInjection/ProjectionMaterializationRuntimeRegistration.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,7 @@ public static IServiceCollection AddProjectionMaterializationRuntimeCore<TContex
5555
sp.GetService<Aevatar.Foundation.Abstractions.TypeSystem.IAgentTypeVerifier>(),
5656
sp.GetService<IStreamPubSubMaintenance>(),
5757
sp.GetService<ILoggerFactory>(),
58-
streams: sp.GetService<IStreamProvider>()),
58+
sp.GetService<IStreamForwardingRegistry>()),
5959
sp.GetService<IProjectionScopeActivationService<ProjectionScopeStatusRuntimeLease>>()));
6060
services.TryAddSingleton<IProjectionScopeReleaseService<TRuntimeLease>>(sp =>
6161
new ProjectionScopeReleaseService<

‎src/Aevatar.CQRS.Projection.Core/DependencyInjection/ProjectionScopeStatusRuntimeRegistration.cs‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -54,8 +54,7 @@ public static IServiceCollection AddProjectionScopeStatusRuntimeCore(this IServi
5454
(_, context) => new ProjectionScopeStatusRuntimeLease(context),
5555
sp.GetService<Aevatar.Foundation.Abstractions.TypeSystem.IAgentTypeVerifier>(),
5656
sp.GetService<IStreamPubSubMaintenance>(),
57-
sp.GetService<ILoggerFactory>(),
58-
streams: sp.GetService<IStreamProvider>()));
57+
sp.GetService<ILoggerFactory>()));
5958
services.TryAddSingleton<IProjectionScopeReleaseService<ProjectionScopeStatusRuntimeLease>>(sp =>
6059
new ProjectionScopeReleaseService<
6160
ProjectionScopeStatusRuntimeLease,

‎src/Aevatar.CQRS.Projection.Core/Orchestration/CommittedStateProjectionActivationHook.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,7 @@ public async Task BeforePublishAsync(CommittedStatePublicationContext context, C
6060

6161
try
6262
{
63-
await _dispatcher.DispatchAsync(plan, ct).ConfigureAwait(false);
63+
await _dispatcher.DispatchAsync(plan, context, ct).ConfigureAwait(false);
6464
}
6565
catch (Exception ex)
6666
{

‎src/Aevatar.CQRS.Projection.Core/Orchestration/ProjectionActivationPlanDispatcher.cs‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
using System.Reflection;
22
using Aevatar.CQRS.Projection.Core.Abstractions;
3+
using Aevatar.Foundation.Abstractions.EventSourcing;
4+
using Microsoft.Extensions.DependencyInjection;
35

46
namespace Aevatar.CQRS.Projection.Core.Orchestration;
57

@@ -33,6 +35,15 @@ public Task DispatchAsync(ProjectionActivationPlan plan, CancellationToken ct =
3335
.Invoke(this, [plan.StartRequest, ct])!;
3436
}
3537

38+
public Task DispatchAsync(
39+
ProjectionActivationPlan plan,
40+
CommittedStatePublicationContext context,
41+
CancellationToken ct = default)
42+
{
43+
ArgumentNullException.ThrowIfNull(context);
44+
return DispatchAsync(plan, ct);
45+
}
46+
3647
// Refactor (iter18/cluster-006):
3748
// Old pattern: command-path projection activation facade with new actor/lifecycle phase
3849
// New principle: committed-state publication hook activates existing projection scopes; no new actor/lifecycle phase

‎src/Aevatar.CQRS.Projection.Core/Orchestration/ProjectionScopeActivationService.cs‎

Lines changed: 39 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -10,9 +10,13 @@ public sealed class ProjectionScopeActivationService<TLease, TContext, TScopeAge
1010
where TContext : class, IProjectionMaterializationContext
1111
where TScopeAgent : IAgent
1212
{
13+
private static readonly TimeSpan RelayReadinessTimeout = TimeSpan.FromSeconds(10);
14+
private static readonly TimeSpan RelayReadinessCheckInterval = TimeSpan.FromMilliseconds(50);
15+
1316
private readonly ProjectionScopeActorRuntime<TScopeAgent> _scopeRuntime;
1417
private readonly Func<ProjectionScopeStartRequest, TContext> _contextFactory;
1518
private readonly Func<ProjectionRuntimeScopeKey, TContext, TLease> _leaseFactory;
19+
private readonly IStreamForwardingRegistry? _forwardingRegistry;
1620

1721
public ProjectionScopeActivationService(
1822
IActorRuntime runtime,
@@ -22,17 +26,17 @@ public ProjectionScopeActivationService(
2226
IAgentTypeVerifier? agentTypeVerifier = null,
2327
IStreamPubSubMaintenance? streamPubSubMaintenance = null,
2428
ILoggerFactory? loggerFactory = null,
25-
IStreamProvider? streams = null)
29+
IStreamForwardingRegistry? forwardingRegistry = null)
2630
{
2731
_scopeRuntime = new ProjectionScopeActorRuntime<TScopeAgent>(
2832
runtime,
2933
dispatchPort,
3034
agentTypeVerifier,
3135
streamPubSubMaintenance,
32-
loggerFactory?.CreateLogger<ProjectionScopeActorRuntime<TScopeAgent>>(),
33-
streams);
36+
loggerFactory?.CreateLogger<ProjectionScopeActorRuntime<TScopeAgent>>());
3437
_contextFactory = contextFactory ?? throw new ArgumentNullException(nameof(contextFactory));
3538
_leaseFactory = leaseFactory ?? throw new ArgumentNullException(nameof(leaseFactory));
39+
_forwardingRegistry = forwardingRegistry;
3640
}
3741

3842
public async Task<TLease> EnsureAsync(
@@ -54,7 +58,6 @@ public async Task<TLease> EnsureAsync(
5458
request.SessionId);
5559

5660
await _scopeRuntime.EnsureExistsAsync(scopeKey, ct).ConfigureAwait(false);
57-
await _scopeRuntime.EnsureObservationRelayAsync(scopeKey, ct).ConfigureAwait(false);
5861
await _scopeRuntime.DispatchAsync(
5962
scopeKey,
6063
new EnsureProjectionScopeCommand
@@ -65,7 +68,39 @@ await _scopeRuntime.DispatchAsync(
6568
Mode = ProjectionScopeModeMapper.ToProto(scopeKey.Mode),
6669
},
6770
ct).ConfigureAwait(false);
71+
await WaitForObservationRelayAsync(scopeKey, ct).ConfigureAwait(false);
6872

6973
return _leaseFactory(scopeKey, context);
7074
}
75+
76+
private async Task WaitForObservationRelayAsync(
77+
ProjectionRuntimeScopeKey scopeKey,
78+
CancellationToken ct)
79+
{
80+
if (_forwardingRegistry == null || string.IsNullOrWhiteSpace(scopeKey.RootActorId))
81+
return;
82+
83+
var targetActorId = ProjectionScopeActorId.Build(scopeKey);
84+
using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
85+
timeout.CancelAfter(RelayReadinessTimeout);
86+
87+
try
88+
{
89+
while (true)
90+
{
91+
var relays = await _forwardingRegistry
92+
.ListBySourceAsync(scopeKey.RootActorId, timeout.Token)
93+
.ConfigureAwait(false);
94+
if (relays.Any(relay => string.Equals(relay.TargetStreamId, targetActorId, StringComparison.Ordinal)))
95+
return;
96+
97+
await Task.Delay(RelayReadinessCheckInterval, timeout.Token).ConfigureAwait(false);
98+
}
99+
}
100+
catch (OperationCanceledException) when (timeout.IsCancellationRequested && !ct.IsCancellationRequested)
101+
{
102+
throw new TimeoutException(
103+
$"Timed out waiting for projection observation relay. root_actor_id={scopeKey.RootActorId} projection_kind={scopeKey.ProjectionKind} session_id={scopeKey.SessionId}");
104+
}
105+
}
71106
}

‎src/Aevatar.CQRS.Projection.Core/Orchestration/ProjectionScopeActorRuntime.cs‎

Lines changed: 1 addition & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -12,23 +12,20 @@ internal sealed class ProjectionScopeActorRuntime<TScopeAgent>
1212
private readonly IActorDispatchPort _dispatchPort;
1313
private readonly IAgentTypeVerifier? _agentTypeVerifier;
1414
private readonly IStreamPubSubMaintenance? _streamPubSubMaintenance;
15-
private readonly IStreamProvider? _streams;
1615
private readonly ILogger<ProjectionScopeActorRuntime<TScopeAgent>> _logger;
1716

1817
public ProjectionScopeActorRuntime(
1918
IActorRuntime runtime,
2019
IActorDispatchPort dispatchPort,
2120
IAgentTypeVerifier? agentTypeVerifier = null,
2221
IStreamPubSubMaintenance? streamPubSubMaintenance = null,
23-
ILogger<ProjectionScopeActorRuntime<TScopeAgent>>? logger = null,
24-
IStreamProvider? streams = null)
22+
ILogger<ProjectionScopeActorRuntime<TScopeAgent>>? logger = null)
2523
{
2624
_runtime = runtime ?? throw new ArgumentNullException(nameof(runtime));
2725
_dispatchPort = dispatchPort ?? throw new ArgumentNullException(nameof(dispatchPort));
2826
_agentTypeVerifier = agentTypeVerifier;
2927
_streamPubSubMaintenance = streamPubSubMaintenance;
3028
_logger = logger ?? NullLogger<ProjectionScopeActorRuntime<TScopeAgent>>.Instance;
31-
_streams = streams;
3229
}
3330

3431
public async Task EnsureExistsAsync(ProjectionRuntimeScopeKey scopeKey, CancellationToken ct)
@@ -105,18 +102,4 @@ public async Task DispatchAsync(
105102
envelope.Route = EnvelopeRouteSemantics.CreateDirect("projection.scope.port", actorId);
106103
_ = await _dispatchPort.DispatchAsync(actorId, envelope, ct).ConfigureAwait(false);
107104
}
108-
109-
public Task EnsureObservationRelayAsync(ProjectionRuntimeScopeKey scopeKey, CancellationToken ct)
110-
{
111-
if (_streams == null || string.IsNullOrWhiteSpace(scopeKey.RootActorId))
112-
return Task.CompletedTask;
113-
114-
return _streams
115-
.GetStream(scopeKey.RootActorId)
116-
.UpsertRelayAsync(
117-
ProjectionScopeObservationRelayBinding.Create(
118-
scopeKey.RootActorId,
119-
ProjectionScopeActorId.Build(scopeKey)),
120-
ct);
121-
}
122105
}

‎src/Aevatar.Scripting.Projection/Orchestration/ScriptingCommittedStateProjectionActivationPlanProvider.cs‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ public IEnumerable<ProjectionActivationPlan> GetPlans(CommittedStatePublicationC
3636
}
3737

3838
if (context.ActorType == typeof(ScriptDefinitionGAgent) &&
39-
context.Published.StateEvent.EventData.Is(ScriptDefinitionUpsertedEvent.Descriptor))
39+
IsDefinitionAuthorityMutation(context.Published.StateEvent.EventData))
4040
{
4141
yield return DurableAuthorityPlan(context.ActorId);
4242
yield break;
@@ -99,6 +99,23 @@ private static ProjectionActivationPlan DurableAuthorityPlan(string actorId) =>
9999
},
100100
};
101101

102+
private static bool IsDefinitionAuthorityMutation(Google.Protobuf.WellKnownTypes.Any eventData)
103+
{
104+
if (eventData.Is(ScriptDefinitionUpsertedEvent.Descriptor))
105+
return true;
106+
107+
if (eventData.Is(ScriptReadModelSchemaDeclaredEvent.Descriptor))
108+
return true;
109+
110+
if (eventData.Is(ScriptReadModelSchemaValidatedEvent.Descriptor))
111+
return true;
112+
113+
if (eventData.Is(ScriptReadModelSchemaActivationFailedEvent.Descriptor))
114+
return true;
115+
116+
return false;
117+
}
118+
102119
private static bool IsCatalogAuthorityMutation(Google.Protobuf.WellKnownTypes.Any eventData)
103120
{
104121
if (eventData.Is(ScriptCatalogRevisionPromotedEvent.Descriptor))

‎test/Aevatar.CQRS.Projection.Core.Tests/CommittedStateProjectionActivationHookTests.cs‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,25 @@ [new StaticPlanProvider(BuildPlan("actor-1", "projection-a", typeof(TestLease)))
2424
activation.Requests[0].Mode.Should().Be(ProjectionRuntimeMode.DurableMaterialization);
2525
}
2626

27+
[Fact]
28+
public async Task BeforePublishAsync_ShouldOnlyEnsureProjectionScopeBeforeNormalPublication()
29+
{
30+
var activation = new RecordingActivationService<TestLease>();
31+
var dispatchPort = new RecordingActorDispatchPort();
32+
var hook = CreateHook(
33+
[new StaticPlanProvider(BuildPlan("actor-1", "projection-a", typeof(TestLease)))],
34+
services =>
35+
{
36+
services.AddSingleton<IProjectionScopeActivationService<TestLease>>(activation);
37+
services.AddSingleton<IActorDispatchPort>(dispatchPort);
38+
});
39+
40+
await hook.BeforePublishAsync(BuildContext(), CancellationToken.None);
41+
42+
activation.Requests.Should().ContainSingle();
43+
dispatchPort.Dispatched.Should().BeEmpty();
44+
}
45+
2746
[Fact]
2847
public async Task BeforePublishAsync_ShouldDeduplicateDuplicatePlansWithinOnePublication()
2948
{
@@ -153,6 +172,18 @@ public Task<TLease> EnsureAsync(ProjectionScopeStartRequest request, Cancellatio
153172
Task.FromException<TLease>(new InvalidOperationException("activation failed"));
154173
}
155174

175+
private sealed class RecordingActorDispatchPort : IActorDispatchPort
176+
{
177+
public List<(string actorId, EventEnvelope envelope)> Dispatched { get; } = [];
178+
179+
public Task<DispatchAdmission> DispatchAsync(string actorId, EventEnvelope envelope, CancellationToken ct = default)
180+
{
181+
ct.ThrowIfCancellationRequested();
182+
Dispatched.Add((actorId, envelope));
183+
return Task.FromResult(DispatchAdmissionFactory.Create(actorId, envelope));
184+
}
185+
}
186+
156187
private sealed class RecordingActivationService<TLease> : IProjectionScopeActivationService<TLease>
157188
where TLease : class, IProjectionRuntimeLease
158189
{

‎test/Aevatar.CQRS.Projection.Core.Tests/ProjectionRuntimeRegistrationTests.cs‎

Lines changed: 104 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
using Aevatar.CQRS.Projection.Core.DependencyInjection;
22
using Aevatar.CQRS.Projection.Core.Orchestration;
3+
using Aevatar.Foundation.Abstractions.Streaming;
34
using FluentAssertions;
5+
using Google.Protobuf;
46
using Google.Protobuf.WellKnownTypes;
57
using Microsoft.Extensions.DependencyInjection;
68

@@ -103,6 +105,44 @@ public async Task AddProjectionMaterializationRuntimeCore_ShouldReleaseSessionSc
103105
.Should().Be("correlation-1");
104106
}
105107

108+
[Fact]
109+
public async Task AddProjectionMaterializationRuntimeCore_ShouldNotWriteObservationRelayFromActivationService()
110+
{
111+
var runtime = new RecordingActorRuntime();
112+
var dispatchPort = new RecordingActorDispatchPort();
113+
var streamProvider = new RecordingStreamProvider();
114+
var services = new ServiceCollection();
115+
services.AddSingleton<IActorRuntime>(runtime);
116+
services.AddSingleton<IActorDispatchPort>(dispatchPort);
117+
services.AddSingleton<IStreamProvider>(streamProvider);
118+
119+
services.AddProjectionMaterializationRuntimeCore<
120+
TestMaterializationContext,
121+
TestMaterializationLease,
122+
ProjectionMaterializationScopeGAgent<TestMaterializationContext>>(
123+
scopeKey => new TestMaterializationContext
124+
{
125+
RootActorId = scopeKey.RootActorId,
126+
ProjectionKind = scopeKey.ProjectionKind,
127+
},
128+
context => new TestMaterializationLease(context));
129+
130+
await using var provider = services.BuildServiceProvider();
131+
var activation = provider.GetRequiredService<IProjectionScopeActivationService<TestMaterializationLease>>();
132+
133+
await activation.EnsureAsync(new ProjectionScopeStartRequest
134+
{
135+
RootActorId = "actor-relay",
136+
ProjectionKind = "projection-relay",
137+
Mode = ProjectionRuntimeMode.DurableMaterialization,
138+
});
139+
140+
streamProvider.Streams.Should().BeEmpty();
141+
dispatchPort.Dispatched.Should().ContainSingle();
142+
var command = dispatchPort.Dispatched[0].command.Payload!.Unpack<EnsureProjectionScopeCommand>();
143+
command.RootActorId.Should().Be("actor-relay");
144+
}
145+
106146
[Fact]
107147
public async Task AddProjectionMaterializationRuntimeCore_ShouldRegisterAttachExistingLeaseLookup()
108148
{
@@ -308,7 +348,7 @@ public async Task AddEventSinkProjectionRuntimeCore_ShouldRegisterAttachExisting
308348

309349
lease.Should().NotBeNull();
310350
lease!.Context.RootActorId.Should().Be("actor-session");
311-
lease.SessionId.Should().Be("session-lookup");
351+
lease.Context.SessionId.Should().Be("session-lookup");
312352
runtime.CreatedActorIds.Should().BeEmpty();
313353
dispatchPort.Dispatched.Should().BeEmpty();
314354
}
@@ -407,6 +447,69 @@ public Task<DispatchAdmission> DispatchAsync(string actorId, EventEnvelope envel
407447
}
408448
}
409449

450+
private sealed class RecordingStreamProvider : IStreamProvider
451+
{
452+
public Dictionary<string, RecordingStream> Streams { get; } = new(StringComparer.Ordinal);
453+
454+
public IStream GetStream(string actorId)
455+
{
456+
if (!Streams.TryGetValue(actorId, out var stream))
457+
{
458+
stream = new RecordingStream(actorId);
459+
Streams[actorId] = stream;
460+
}
461+
462+
return stream;
463+
}
464+
}
465+
466+
private sealed class RecordingStream(string streamId) : IStream
467+
{
468+
public string StreamId { get; } = streamId;
469+
470+
public List<StreamForwardingBinding> UpsertedRelays { get; } = [];
471+
472+
public Task ProduceAsync<T>(T message, CancellationToken ct = default)
473+
where T : IMessage
474+
{
475+
ct.ThrowIfCancellationRequested();
476+
return Task.CompletedTask;
477+
}
478+
479+
public Task<IAsyncDisposable> SubscribeAsync<T>(Func<T, Task> handler, CancellationToken ct = default)
480+
where T : IMessage, new()
481+
{
482+
ct.ThrowIfCancellationRequested();
483+
_ = handler;
484+
return Task.FromResult<IAsyncDisposable>(new NoOpSubscription());
485+
}
486+
487+
public Task UpsertRelayAsync(StreamForwardingBinding binding, CancellationToken ct = default)
488+
{
489+
ct.ThrowIfCancellationRequested();
490+
UpsertedRelays.Add(binding);
491+
return Task.CompletedTask;
492+
}
493+
494+
public Task RemoveRelayAsync(string targetStreamId, CancellationToken ct = default)
495+
{
496+
ct.ThrowIfCancellationRequested();
497+
_ = targetStreamId;
498+
return Task.CompletedTask;
499+
}
500+
501+
public Task<IReadOnlyList<StreamForwardingBinding>> ListRelaysAsync(CancellationToken ct = default)
502+
{
503+
ct.ThrowIfCancellationRequested();
504+
return Task.FromResult<IReadOnlyList<StreamForwardingBinding>>(UpsertedRelays);
505+
}
506+
507+
private sealed class NoOpSubscription : IAsyncDisposable
508+
{
509+
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
510+
}
511+
}
512+
410513
private sealed class RecordingActor : IActor
411514
{
412515
public RecordingActor(string id)

0 commit comments

Comments
 (0)