feat(persistent-subscriptions): allow parked message truncation (#437)

- give operators a safe way to discard parked messages without
redelivering them
- keep persistent subscription parked-message metrics aligned with
operator cleanup actions
- expose the cleanup path through gRPC so automation does not need to
mutate parked streams directly

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>

Author: yordis

proto.lock

@@ -4326,6 +4326,51 @@
           {
             "name": "ReplayParkedResp"
           },
+          {
+            "name": "TruncateParkedReq",
+            "fields": [
+              {
+                "id": 1,
+                "name": "options",
+                "type": "Options"
+              }
+            ],
+            "messages": [
+              {
+                "name": "Options",
+                "fields": [
+                  {
+                    "id": 1,
+                    "name": "group_name",
+                    "type": "string"
+                  },
+                  {
+                    "id": 2,
+                    "name": "stream_identifier",
+                    "type": "event_store.client.StreamIdentifier"
+                  },
+                  {
+                    "id": 3,
+                    "name": "all",
+                    "type": "event_store.client.Empty"
+                  },
+                  {
+                    "id": 4,
+                    "name": "stop_at",
+                    "type": "int64"
+                  },
+                  {
+                    "id": 5,
+                    "name": "no_limit",
+                    "type": "event_store.client.Empty"
+                  }
+                ]
+              }
+            ]
+          },
+          {
+            "name": "TruncateParkedResp"
+          },
           {
             "name": "ListReq",
             "fields": [
@@ -4441,6 +4486,11 @@
                 "in_type": "ReplayParkedReq",
                 "out_type": "ReplayParkedResp"
               },
+              {
+                "name": "TruncateParked",
+                "in_type": "TruncateParkedReq",
+                "out_type": "TruncateParkedResp"
+              },
               {
                 "name": "List",
                 "in_type": "ListReq",
@@ -7280,4 +7330,4 @@
       }
     }
   ]
-}
+}

src/EventStore.Core.Tests/Services/PersistentSubscription/PersistentSubscriptionMessageParkerTests.cs

@@ -360,6 +360,63 @@ public async Task should_have_one_parked_message()
 	}
 
 
+	[TestFixture(typeof(LogFormat.V2), typeof(string))]
+	public class given_messages_are_parked_and_then_truncated<TLogFormat, TStreamId> : TestFixtureWithExistingEvents<TLogFormat, TStreamId>
+	{
+		private PersistentSubscriptionMessageParker _messageParker;
+		private string _streamId = Guid.NewGuid().ToString();
+		private TaskCompletionSource<bool> _parked;
+		private TaskCompletionSource<bool> _done = new TaskCompletionSource<bool>();
+
+		protected override void Given()
+		{
+			base.Given();
+
+			AllWritesSucceed();
+
+			_parked = new TaskCompletionSource<bool>();
+			_messageParker = new PersistentSubscriptionMessageParker(_streamId, _ioDispatcher);
+			_messageParker.BeginParkMessage(CreateResolvedEvent(0, 0), "testing", (_, __) =>
+			{
+				_messageParker.BeginParkMessage(CreateResolvedEvent(1, 100), "testing", (_, __) =>
+				{
+					_parked.SetResult(true);
+				});
+			});
+		}
+
+		[Test]
+		public async Task should_have_no_parked_messages()
+		{
+			await _parked.Task;
+			_messageParker.BeginMarkParkedMessagesTruncated(2, _ =>
+			{
+				Assert.Zero(_messageParker.ParkedMessageCount);
+				Assert.Null(_messageParker.GetOldestParkedMessage);
+				Assert.Zero(_messageParker.ParkedMessageReplays);
+				_done.TrySetResult(true);
+			});
+			await _done.Task.WithTimeout();
+		}
+
+		[Test]
+		public async Task should_not_lower_the_truncate_before_watermark()
+		{
+			await _parked.Task;
+			_messageParker.BeginMarkParkedMessagesTruncated(2, _ =>
+			{
+				_messageParker.BeginMarkParkedMessagesTruncated(0, __ =>
+				{
+					Assert.Zero(_messageParker.ParkedMessageCount);
+					Assert.Null(_messageParker.GetOldestParkedMessage);
+					_done.TrySetResult(true);
+				});
+			});
+			await _done.Task.WithTimeout();
+		}
+	}
+
+
 	[TestFixture(typeof(LogFormat.V2), typeof(string))]
 	public class given_read_backwards_fails_when_getting_stats<TLogFormat, TStreamId> : TestFixtureWithExistingEvents<TLogFormat, TStreamId>
 	{

src/EventStore.Core.Tests/Services/PersistentSubscription/PersistentSubscriptionServiceNotReadyTests.cs

@@ -172,6 +172,18 @@ public void replay_parked_replies_not_ready()
 		AssertNotReady(envelope, correlationId);
 	}
 
+	[Test]
+	public void truncate_parked_replies_not_ready()
+	{
+		var envelope = new FakeEnvelope();
+		var correlationId = Guid.NewGuid();
+
+		_sut.Handle(new ClientMessage.TruncateParkedMessages(
+			Guid.NewGuid(), correlationId, envelope, "stream", "group", null, ClaimsPrincipal.Current));
+
+		AssertNotReady(envelope, correlationId);
+	}
+
 	private static void AssertNotReady(FakeEnvelope envelope, Guid correlationId)
 	{
 		Assert.That(envelope.Replies, Has.Count.EqualTo(1));

src/EventStore.Core.Tests/Services/PersistentSubscription/PersistentSubscriptionTests.cs

@@ -2980,6 +2980,16 @@ public void RecordParkMessageRequest(ParkReason parkReason)
 		}
 	}
 
+	public void RecordParkedMessageReplay()
+	{
+		ParkedMessageReplays++;
+	}
+
+	public void RecordParkedMessageTruncate()
+	{
+		ParkedMessageTruncates++;
+	}
+
 	public void BeginParkMessage(ResolvedEvent ev, string reason,
 		Action<ResolvedEvent, OperationResult> completed)
 	{
@@ -2988,23 +2998,30 @@ public void BeginParkMessage(ResolvedEvent ev, string reason,
 		_parkMessageCompleted = completed;
 	}
 
-	public void BeginReadEndSequence(Action<long?> completed)
+	public void BeginReadEndSequence(Action<ParkedStreamEndReadResult> completed)
 	{
 		BeginReadEndSequenceCount++;
-		ParkedMessageReplays++;
 		if (_lastParkedEventNumber == -1)
 		{
-			completed(null); //NoStream
+			completed(ParkedStreamEndReadResult.Success(null)); //NoStream
 		}
 		else
 		{
-			completed(_lastParkedEventNumber);
+			completed(ParkedStreamEndReadResult.Success(_lastParkedEventNumber));
 		}
 	}
 
 	public void BeginMarkParkedMessagesReprocessed(long sequence, DateTime? dateTime, bool updateOldestParkedMessage)
 	{
 		MarkedAsProcessed = sequence;
+		_lastTruncateBefore = sequence;
+	}
+
+	public void BeginMarkParkedMessagesTruncated(long sequence, Action<OperationResult> completed)
+	{
+		MarkedAsProcessed = sequence;
+		_lastTruncateBefore = sequence;
+		completed?.Invoke(OperationResult.Success);
 	}
 
 	public void BeginDelete(Action<IPersistentSubscriptionMessageParker> completed)
@@ -3031,6 +3048,7 @@ public void BeginLoadStats(Action completed)
 	public long ParkedDueToClientNak { get; private set; }
 	public long ParkedDueToMaxRetries { get; private set; }
 	public long ParkedMessageReplays { get; private set; }
+	public long ParkedMessageTruncates { get; private set; }
 }