From 75b79a7daa04ff737de6f9d5b525eaa922cf9183 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 6 Jul 2026 16:34:38 -0400 Subject: [PATCH] fix(projections): reject oversized definitions Signed-off-by: Yordis Prieto --- ..._a_projection_with_oversized_definition.cs | 67 ++++++++++++++++ ..._a_projection_with_oversized_definition.cs | 77 +++++++++++++++++++ .../Messages/ProjectionManagementMessage.cs | 9 +++ .../Grpc/ProjectionManagement.Abort.cs | 3 + .../Grpc/ProjectionManagement.Create.cs | 12 ++- .../Grpc/ProjectionManagement.Delete.cs | 3 + .../Grpc/ProjectionManagement.Disable.cs | 3 + .../Grpc/ProjectionManagement.Enable.cs | 3 + .../Grpc/ProjectionManagement.Reset.cs | 3 + .../Grpc/ProjectionManagement.Update.cs | 3 + .../Services/Grpc/ProjectionManagement.cs | 13 ++++ .../Services/Management/ManagedProjection.cs | 57 ++++++++++++-- 12 files changed, 242 insertions(+), 11 deletions(-) create mode 100644 src/EventStore.Projections.Core.Tests/Services/projections_manager/when_posting_a_projection_with_oversized_definition.cs create mode 100644 src/EventStore.Projections.Core.Tests/Services/projections_manager/when_updating_a_projection_with_oversized_definition.cs diff --git a/src/EventStore.Projections.Core.Tests/Services/projections_manager/when_posting_a_projection_with_oversized_definition.cs b/src/EventStore.Projections.Core.Tests/Services/projections_manager/when_posting_a_projection_with_oversized_definition.cs new file mode 100644 index 0000000000..e46c49bd80 --- /dev/null +++ b/src/EventStore.Projections.Core.Tests/Services/projections_manager/when_posting_a_projection_with_oversized_definition.cs @@ -0,0 +1,67 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using EventStore.Core.Messages; +using EventStore.Core.Tests; +using EventStore.Core.TransactionLog.Chunks; +using EventStore.Projections.Core.Messages; +using EventStore.Projections.Core.Services; +using EventStore.Projections.Core.Services.Management; +using EventStore.Projections.Core.Services.Processing; +using NUnit.Framework; + +namespace EventStore.Projections.Core.Tests.Services.projections_manager; + +[TestFixture(typeof(LogFormat.V2), typeof(string))] +public class when_posting_a_projection_with_oversized_definition : TestFixtureWithProjectionCoreAndManagementServices +{ + private string _projectionName; + private readonly string _oversizedQuery = new('a', TFConsts.EffectiveMaxLogRecordSize); + + protected override void Given() + { + _projectionName = "test-projection"; + NoStream("$projections-test-projection"); + NoStream("$projections-test-projection-result"); + NoStream("$projections-test-projection-order"); + AllWritesToSucceed("$projections-test-projection-order"); + NoStream("$projections-test-projection-checkpoint"); + AllWritesSucceed(); + } + + protected override IEnumerable When() + { + yield return new ProjectionSubsystemMessage.StartComponents(Guid.NewGuid()); + yield return + new ProjectionManagementMessage.Command.Post( + _bus, ProjectionMode.Continuous, _projectionName, + ProjectionManagementMessage.RunAs.System, "JS", _oversizedQuery, + enabled: true, checkpointsEnabled: true, emitEnabled: true, trackEmittedStreams: true); + } + + [Test, Category("v8")] + public void the_operation_fails_as_record_too_large() + { + var failure = _consumer.HandledMessages.OfType().Single(); + StringAssert.Contains("exceeds the maximum", failure.Reason); + } + + [Test, Category("v8")] + public void no_definition_event_is_written() + { + var persistedStateStream = ProjectionNamesBuilder.ProjectionsStreamPrefix + _projectionName; + Assert.IsFalse(_consumer.HandledMessages.OfType() + .Any(x => x.EventStreamId == persistedStateStream && + x.Events[0].EventType == ProjectionEventTypes.ProjectionUpdated)); + } + + [Test, Category("v8")] + public void the_projection_is_faulted() + { + _manager.Handle(new ProjectionManagementMessage.Command.GetStatistics(_bus, null, _projectionName, true)); + Assert.AreEqual( + ManagedProjectionState.Faulted, + _consumer.HandledMessages.OfType().Last().Projections.Single() + .LeaderStatus); + } +} diff --git a/src/EventStore.Projections.Core.Tests/Services/projections_manager/when_updating_a_projection_with_oversized_definition.cs b/src/EventStore.Projections.Core.Tests/Services/projections_manager/when_updating_a_projection_with_oversized_definition.cs new file mode 100644 index 0000000000..c869cb3e4d --- /dev/null +++ b/src/EventStore.Projections.Core.Tests/Services/projections_manager/when_updating_a_projection_with_oversized_definition.cs @@ -0,0 +1,77 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using EventStore.Core.Messages; +using EventStore.Core.Tests; +using EventStore.Core.TransactionLog.Chunks; +using EventStore.Projections.Core.Messages; +using EventStore.Projections.Core.Services; +using EventStore.Projections.Core.Services.Management; +using EventStore.Projections.Core.Services.Processing; +using NUnit.Framework; + +namespace EventStore.Projections.Core.Tests.Services.projections_manager; + +[TestFixture(typeof(LogFormat.V2), typeof(string))] +public class when_updating_a_projection_with_oversized_definition : TestFixtureWithProjectionCoreAndManagementServices +{ + private string _projectionName; + private readonly string _oversizedQuery = new('a', TFConsts.EffectiveMaxLogRecordSize); + + protected override void Given() + { + _projectionName = "test-projection"; + NoStream("$projections-test-projection"); + NoStream("$projections-test-projection-result"); + NoStream("$projections-test-projection-order"); + AllWritesToSucceed("$projections-test-projection-order"); + NoStream("$projections-test-projection-checkpoint"); + AllWritesSucceed(); + } + + protected override IEnumerable When() + { + yield return new ProjectionSubsystemMessage.StartComponents(Guid.NewGuid()); + yield return + new ProjectionManagementMessage.Command.Post( + _bus, ProjectionMode.Continuous, _projectionName, + ProjectionManagementMessage.RunAs.System, "JS", @"fromAll().when({$any:function(s,e){return s;}});", + enabled: true, checkpointsEnabled: true, emitEnabled: true, trackEmittedStreams: true); + yield return + new ProjectionManagementMessage.Command.UpdateQuery( + _bus, _projectionName, ProjectionManagementMessage.RunAs.System, + _oversizedQuery, emitEnabled: null); + } + + [Test, Category("v8")] + public void the_operation_fails_as_record_too_large() + { + var failure = _consumer.HandledMessages.OfType().Single(); + StringAssert.Contains("exceeds the maximum", failure.Reason); + } + + [Test, Category("v8")] + public void no_second_definition_event_is_written() + { + var persistedStateStream = ProjectionNamesBuilder.ProjectionsStreamPrefix + _projectionName; + Assert.AreEqual(1, _consumer.HandledMessages.OfType() + .Count(x => x.EventStreamId == persistedStateStream && + x.Events[0].EventType == ProjectionEventTypes.ProjectionUpdated)); + } + + [Test, Category("v8")] + public void the_projection_is_faulted() + { + _manager.Handle(new ProjectionManagementMessage.Command.GetStatistics(_bus, null, _projectionName, true)); + Assert.AreEqual( + ManagedProjectionState.Faulted, + _consumer.HandledMessages.OfType().Last().Projections.Single() + .LeaderStatus); + } + + [Test, Category("v8")] + public void the_core_projection_is_disposed() + { + Assert.AreEqual(1, _consumer.HandledMessages.OfType().Count()); + } +} diff --git a/src/EventStore.Projections.Core/Messages/ProjectionManagementMessage.cs b/src/EventStore.Projections.Core/Messages/ProjectionManagementMessage.cs index 0dd4e6abdb..b76f4097b4 100644 --- a/src/EventStore.Projections.Core/Messages/ProjectionManagementMessage.cs +++ b/src/EventStore.Projections.Core/Messages/ProjectionManagementMessage.cs @@ -639,6 +639,15 @@ public Conflict(string reason) } } + [DerivedMessage(ProjectionMessage.Management)] + public partial class RecordTooLarge : OperationFailed + { + public RecordTooLarge(string reason) + : base(reason) + { + } + } + public sealed class RunAs { private readonly ClaimsPrincipal _runAs; diff --git a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Abort.cs b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Abort.cs index cc5d9ef23f..49a8a61a76 100644 --- a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Abort.cs +++ b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Abort.cs @@ -44,6 +44,9 @@ void OnMessage(Message message) case ProjectionManagementMessage.NotFound: abortSource.TrySetException(ProjectionNotFound(name)); break; + case ProjectionManagementMessage.OperationFailed failed: + abortSource.TrySetException(MapFailure(failed)); + break; default: abortSource.TrySetException(UnknownMessage(message)); break; diff --git a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Create.cs b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Create.cs index 727891d472..97658491dc 100644 --- a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Create.cs +++ b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Create.cs @@ -82,13 +82,19 @@ public override async Task Create(CreateReq request, ServerCallConte void OnMessage(Message message) { - if (message is not ProjectionManagementMessage.Updated) + if (message is ProjectionManagementMessage.Updated) { - createdSource.TrySetException(UnknownMessage(message)); + createdSource.TrySetResult(true); return; } - createdSource.TrySetResult(true); + if (message is ProjectionManagementMessage.OperationFailed failed) + { + createdSource.TrySetException(MapFailure(failed)); + return; + } + + createdSource.TrySetException(UnknownMessage(message)); } } } diff --git a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Delete.cs b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Delete.cs index ecd9727e80..87bb33968e 100644 --- a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Delete.cs +++ b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Delete.cs @@ -46,6 +46,9 @@ void OnMessage(Message message) case ProjectionManagementMessage.NotFound: deletedSource.TrySetException(ProjectionNotFound(name)); break; + case ProjectionManagementMessage.OperationFailed failed: + deletedSource.TrySetException(MapFailure(failed)); + break; default: deletedSource.TrySetException(UnknownMessage(message)); break; diff --git a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Disable.cs b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Disable.cs index bb99061378..144436f564 100644 --- a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Disable.cs +++ b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Disable.cs @@ -45,6 +45,9 @@ void OnMessage(Message message) case ProjectionManagementMessage.NotFound: disableSource.TrySetException(ProjectionNotFound(name)); break; + case ProjectionManagementMessage.OperationFailed failed: + disableSource.TrySetException(MapFailure(failed)); + break; default: disableSource.TrySetException(UnknownMessage(message)); break; diff --git a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Enable.cs b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Enable.cs index de6814423e..e4efe5da02 100644 --- a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Enable.cs +++ b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Enable.cs @@ -43,6 +43,9 @@ void OnMessage(Message message) case ProjectionManagementMessage.NotFound: enableSource.TrySetException(ProjectionNotFound(name)); break; + case ProjectionManagementMessage.OperationFailed failed: + enableSource.TrySetException(MapFailure(failed)); + break; default: enableSource.TrySetException(UnknownMessage(message)); break; diff --git a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Reset.cs b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Reset.cs index cc383e6e9b..876bff81e1 100644 --- a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Reset.cs +++ b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Reset.cs @@ -43,6 +43,9 @@ void OnMessage(Message message) case ProjectionManagementMessage.NotFound: resetSource.TrySetException(ProjectionNotFound(name)); break; + case ProjectionManagementMessage.OperationFailed failed: + resetSource.TrySetException(MapFailure(failed)); + break; default: resetSource.TrySetException(UnknownMessage(message)); break; diff --git a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Update.cs b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Update.cs index 652ca76a30..be3379d76e 100644 --- a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Update.cs +++ b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Update.cs @@ -54,6 +54,9 @@ void OnMessage(Message message) case ProjectionManagementMessage.NotFound: updatedSource.TrySetException(ProjectionNotFound(name)); break; + case ProjectionManagementMessage.OperationFailed failed: + updatedSource.TrySetException(MapFailure(failed)); + break; default: updatedSource.TrySetException(UnknownMessage(message)); break; diff --git a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.cs b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.cs index 60029750a8..6930d48ed2 100644 --- a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.cs +++ b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.cs @@ -6,6 +6,7 @@ using EventStore.Core.Messaging; using EventStore.Core.Services.Transport.Grpc; using EventStore.Plugins.Authorization; +using EventStore.Projections.Core.Messages; using Google.Protobuf.WellKnownTypes; using Grpc.Core; @@ -54,6 +55,18 @@ private static Exception InvalidSubsystemRestart(string state) => private static Exception ProjectionNotFound(string name) => new RpcException(new Status(StatusCode.NotFound, $"Projection '{name}' not found")); + private static Exception MapFailure(ProjectionManagementMessage.OperationFailed failed) => + failed switch + { + ProjectionManagementMessage.RecordTooLarge => new RpcException( + new Status(StatusCode.InvalidArgument, failed.Reason)), + ProjectionManagementMessage.Conflict => new RpcException( + new Status(StatusCode.AlreadyExists, failed.Reason)), + ProjectionManagementMessage.NotAuthorized => new RpcException( + new Status(StatusCode.PermissionDenied, failed.Reason)), + _ => new RpcException(new Status(StatusCode.FailedPrecondition, failed.Reason)) + }; + private static Value GetProtoValue(JsonElement element) => element.ValueKind switch { diff --git a/src/EventStore.Projections.Core/Services/Management/ManagedProjection.cs b/src/EventStore.Projections.Core/Services/Management/ManagedProjection.cs index 6d857e86bf..62a1cd74e1 100644 --- a/src/EventStore.Projections.Core/Services/Management/ManagedProjection.cs +++ b/src/EventStore.Projections.Core/Services/Management/ManagedProjection.cs @@ -11,6 +11,7 @@ using EventStore.Core.Messaging; using EventStore.Core.Services.TimerService; using EventStore.Core.Services.UserManagement; +using EventStore.Core.TransactionLog.Chunks; using EventStore.Projections.Core.Common; using EventStore.Projections.Core.Messages; using EventStore.Projections.Core.Services.Management.ManagedProjectionStates; @@ -573,8 +574,14 @@ public void Handle(ProjectionManagementMessage.Command.UpdateConfig message) UpdateProjectionVersion(); _pendingWritePersistedState = true; - WritePersistedState(CreatePersistedStateEvent(Guid.NewGuid(), PersistedProjectionState, - ProjectionNamesBuilder.ProjectionsStreamPrefix + _name)); + if (!TryCreatePersistedStateEvent(Guid.NewGuid(), PersistedProjectionState, + ProjectionNamesBuilder.ProjectionsStreamPrefix + _name, out var persistedStateEvent, out var recordSize)) + { + FailPersistedStateWrite(recordSize, message.Envelope); + return; + } + + WritePersistedState(persistedStateEvent); message.Envelope.ReplyWith(new ProjectionManagementMessage.Updated(message.Name)); } @@ -845,8 +852,15 @@ internal void WriteStartOrLoadStopped() { if (_pendingWritePersistedState) { - WritePersistedState(CreatePersistedStateEvent(Guid.NewGuid(), PersistedProjectionState, - ProjectionNamesBuilder.ProjectionsStreamPrefix + _name)); + if (!TryCreatePersistedStateEvent(Guid.NewGuid(), PersistedProjectionState, + ProjectionNamesBuilder.ProjectionsStreamPrefix + _name, out var persistedStateEvent, out var recordSize)) + { + FailPersistedStateWrite(recordSize, _lastReplyEnvelope); + _lastReplyEnvelope = null; + return; + } + + WritePersistedState(persistedStateEvent); } else { @@ -859,14 +873,41 @@ internal void StartCompleted() Reply(); } - private ClientMessage.WriteEvents CreatePersistedStateEvent(Guid correlationId, PersistedState persistedState, - string eventStreamId) + private bool TryCreatePersistedStateEvent(Guid correlationId, PersistedState persistedState, string eventStreamId, + out ClientMessage.WriteEvents persistedStateEvent, out int recordSize) { - return new ClientMessage.WriteEvents( + var data = persistedState.ToJsonBytes(); + recordSize = Event.SizeOnDisk(ProjectionEventTypes.ProjectionUpdated, data, Empty.ByteArray); + if (recordSize > TFConsts.EffectiveMaxLogRecordSize) + { + persistedStateEvent = null; + return false; + } + + persistedStateEvent = new ClientMessage.WriteEvents( correlationId, correlationId, _writeDispatcher.Envelope, true, eventStreamId, ExpectedVersion.Any, - new Event(Guid.NewGuid(), ProjectionEventTypes.ProjectionUpdated, true, persistedState.ToJsonBytes(), + new Event(Guid.NewGuid(), ProjectionEventTypes.ProjectionUpdated, true, data, Empty.ByteArray), SystemAccounts.System); + return true; + } + + private void FailPersistedStateWrite(int recordSize, IEnvelope replyEnvelope) + { + _pendingWritePersistedState = false; + var reason = + $"Projection '{_name}' definition record size ({recordSize} bytes) exceeds the maximum of {TFConsts.EffectiveMaxLogRecordSize} bytes."; + if (Created) + { + DisposeCoreProjection(); + } + + if (_state != ManagedProjectionState.Faulted) + { + Fault(reason); + } + + replyEnvelope?.ReplyWith(new ProjectionManagementMessage.RecordTooLarge(reason)); } private void WritePersistedState(ClientMessage.WriteEvents persistedStateEvent)