fix(projections): reject oversized definitions

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<TLogFormat, TStreamId> : TestFixtureWithProjectionCoreAndManagementServices<TLogFormat, TStreamId>
+{
+	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<WhenStep> 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<ProjectionManagementMessage.RecordTooLarge>().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<ClientMessage.WriteEvents>()
+			.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<ProjectionManagementMessage.Statistics>().Last().Projections.Single()
+				.LeaderStatus);
+	}
+}

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<TLogFormat, TStreamId> : TestFixtureWithProjectionCoreAndManagementServices<TLogFormat, TStreamId>
+{
+	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<WhenStep> 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<ProjectionManagementMessage.RecordTooLarge>().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<ClientMessage.WriteEvents>()
+			.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<ProjectionManagementMessage.Statistics>().Last().Projections.Single()
+				.LeaderStatus);
+	}
+
+	[Test, Category("v8")]
+	public void the_core_projection_is_disposed()
+	{
+		Assert.AreEqual(1, _consumer.HandledMessages.OfType<CoreProjectionManagementMessage.Dispose>().Count());
+	}
+}

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;

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<ProjectionManagementMessage.Updated>(message));
 					break;

src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Create.cs

@@ -82,13 +82,19 @@ public override async Task<CreateResp> Create(CreateReq request, ServerCallConte
 
 		void OnMessage(Message message)
 		{
-			if (message is not ProjectionManagementMessage.Updated)
+			if (message is ProjectionManagementMessage.Updated)
 			{
-				createdSource.TrySetException(UnknownMessage<ProjectionManagementMessage.Updated>(message));
+				createdSource.TrySetResult(true);
 				return;
 			}
 
-			createdSource.TrySetResult(true);
+			if (message is ProjectionManagementMessage.OperationFailed failed)
+			{
+				createdSource.TrySetException(MapFailure(failed));
+				return;
+			}
+
+			createdSource.TrySetException(UnknownMessage<ProjectionManagementMessage.Updated>(message));
 		}
 	}
 }

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<ProjectionManagementMessage.Updated>(message));
 					break;

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<ProjectionManagementMessage.Updated>(message));
 					break;

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<ProjectionManagementMessage.Updated>(message));
 					break;

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<ProjectionManagementMessage.Updated>(message));
 					break;

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<ProjectionManagementMessage.Updated>(message));
 					break;

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
 			{

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)