Merge pull request #2824 from EventStore/prevent-checkpoint-at-same-position

Prevent a checkpoint from being emitted at same position twice

Author: hayley-jean

src/EventStore.Projections.Core.Tests/Services/projection_subscription/when_handling_events_with_different_event_filters.cs

@@ -21,12 +21,8 @@ namespace EventStore.Projections.Core.Tests.Services.projection_subscription {
 	[TestFixture("good-stream", "good-event-type", "good-stream", "bad-event-type", true, false)]
 	[TestFixture("good-stream", "good-event-type", "good-stream", "bad-event-type", false, true)]
 	[TestFixture("good-stream", "good-event-type", "good-stream", "bad-event-type", false, false)]
-	[TestFixture("good-stream", "good-event-type", "bad-stream", "good-event-type", true, true)]
-	[TestFixture("good-stream", "good-event-type", "bad-stream", "good-event-type", true, false)]
 	[TestFixture("good-stream", "good-event-type", "bad-stream", "good-event-type", false, true)]
 	[TestFixture("good-stream", "good-event-type", "bad-stream", "good-event-type", false, false)]
-	[TestFixture("good-stream", "good-event-type", "bad-stream", "bad-event-type", true, true)]
-	[TestFixture("good-stream", "good-event-type", "bad-stream", "bad-event-type", true, false)]
 	[TestFixture("good-stream", "good-event-type", "bad-stream", "bad-event-type", false, true)]
 	[TestFixture("good-stream", "good-event-type", "bad-stream", "bad-event-type", false, false)]
 	public class

src/EventStore.Projections.Core/Services/Processing/ReaderSubscriptionBase.cs

@@ -30,6 +30,7 @@ public class ReaderSubscriptionBase {
 		private DateTime _lastCheckpointTime = DateTime.MinValue;
 		private bool _enableContentTypeValidation;
 		private ILogger _logger;
+		private CheckpointTag _lastCheckpointTag;
 
 		protected ReaderSubscriptionBase(
 			IPublisher publisher,
@@ -65,6 +66,7 @@ protected ReaderSubscriptionBase(
 			_positionTagger = readerStrategy.PositionTagger;
 			_positionTracker = new PositionTracker(_positionTagger);
 			_positionTracker.UpdateByCheckpointTagInitial(@from);
+			_lastCheckpointTag = _positionTracker.LastTag;
 			_enableContentTypeValidation = enableContentTypeValidation;
 			_logger = Serilog.Log.ForContext<ReaderSubscriptionBase>();
 		}
@@ -125,7 +127,8 @@ protected void ProcessOne(ReaderSubscriptionMessage.CommittedEventDistributed me
 				_eventsSinceLastCheckpointSuggestedOrStart++;
 				if (_checkpointProcessedEventsThreshold > 0
 				    && timeDifference > _checkpointAfter
-				    && _eventsSinceLastCheckpointSuggestedOrStart >= _checkpointProcessedEventsThreshold)
+				    && _eventsSinceLastCheckpointSuggestedOrStart >= _checkpointProcessedEventsThreshold
+				    && _lastCheckpointTag != _positionTracker.LastTag)
 					SuggestCheckpoint(message);
 				if (_stopAfterNEvents > 0 && _eventsSinceLastCheckpointSuggestedOrStart >= _stopAfterNEvents)
 					NEventsReached();
@@ -134,7 +137,8 @@ protected void ProcessOne(ReaderSubscriptionMessage.CommittedEventDistributed me
 				    && timeDifference > _checkpointAfter
 				    && (_lastPassedOrCheckpointedEventPosition != null
 				        && message.Data.Position.PreparePosition - _lastPassedOrCheckpointedEventPosition.Value
-				        > _checkpointUnhandledBytesThreshold))
+				        > _checkpointUnhandledBytesThreshold)
+				    && _lastCheckpointTag != _positionTracker.LastTag)
 					SuggestCheckpoint(message);
 				else if (progressChanged)
 					PublishProgress(roundedProgress);
@@ -178,6 +182,7 @@ private void PublishStartingAt(long startingLastCommitPosition) {
 
 		private void SuggestCheckpoint(ReaderSubscriptionMessage.CommittedEventDistributed message) {
 			_lastPassedOrCheckpointedEventPosition = message.Data.Position.PreparePosition;
+			_lastCheckpointTag = _positionTracker.LastTag;
 			_publisher.Publish(
 				new EventReaderSubscriptionMessage.CheckpointSuggested(
 					_subscriptionId, _positionTracker.LastTag, message.Progress,