Fixed logic on how to get current events by tag in batches

src/contrib/persistence/Akka.Persistence.Query.Sql/EventsByTagPublisher.cs

@@ -194,7 +194,7 @@ protected override void ReceiveRecoverySuccess(long highestSequenceNr)
             if (highestSequenceNr < ToOffset)
                 _toOffset = highestSequenceNr;
 
-            if (Buffer.IsEmpty && CurrentOffset > ToOffset)
+            if (Buffer.IsEmpty && CurrentOffset >= ToOffset)
                 OnCompleteThenStop();
             else
                 Self.Tell(EventsByTagPublisher.Continue.Instance);

src/contrib/persistence/Akka.Persistence.Sql.Common/Journal/QueryExecutor.cs

@@ -76,7 +76,7 @@ public interface IJournalQueryExecutor
         /// <param name="persistenceId">TBD</param>
         /// <returns>TBD</returns>
         Task<long> SelectHighestSequenceNrAsync(DbConnection connection, CancellationToken cancellationToken, string persistenceId);
-
+        
         /// <summary>
         /// Asynchronously inserts a collection of events and theirs tags into a journal table.
         /// </summary>
@@ -344,6 +344,12 @@ DELETE FROM {Configuration.FullJournalTableName}
                 AND e.{Configuration.SequenceNrColumnName} BETWEEN @FromSequenceNr AND @ToSequenceNr
                 ORDER BY {Configuration.SequenceNrColumnName} ASC;";
 
+            HighestTagOrderingSql =
+                $@"
+                SELECT MAX(e.{Configuration.OrderingColumnName}) as Ordering
+                FROM {Configuration.FullJournalTableName} e
+                WHERE e.{Configuration.OrderingColumnName} > @Ordering AND e.{Configuration.TagsColumnName} LIKE @Tag";
+
             ByTagSql =
                 $@"
                 SELECT {allEventColumnNames}, e.{Configuration.OrderingColumnName} as Ordering
@@ -399,6 +405,10 @@ WHERE e.{Configuration.OrderingColumnName} > @Ordering AND e.{Configuration.Tags
         /// </summary>
         protected virtual string ByPersistenceIdSql { get; }
         /// <summary>
+        /// Query to return the highest ordering number for a tag
+        /// </summary>
+        protected virtual string HighestTagOrderingSql { get; }
+        /// <summary>
         /// TBD
         /// </summary>
         protected virtual string ByTagSql { get; }
@@ -526,18 +536,22 @@ public virtual async Task<long> SelectByTagAsync(DbConnection connection, Cancel
 
                 using (var reader = await command.ExecuteReaderAsync(commandBehavior, cancellationToken))
                 {
-                    var maxSequenceNr = 0L;
                     while (await reader.ReadAsync(cancellationToken))
                     {
                         var persistent = ReadEvent(reader);
                         var ordering = reader.GetInt64(OrderingIndex);
-                        maxSequenceNr = Math.Max(maxSequenceNr, persistent.SequenceNr);
                         callback(new ReplayedTaggedMessage(persistent, tag, ordering));
                     }
-
-                    return maxSequenceNr;
                 }
             }
+
+            using (var command = GetCommand(connection, HighestTagOrderingSql))
+            {
+                AddParameter(command, "@Tag", DbType.String, "%;" + tag + ";%");
+                AddParameter(command, "@Ordering", DbType.Int64, fromOffset);
+                var maxOrdering = (await command.ExecuteScalarAsync(cancellationToken)) as long? ?? 0L;
+                return maxOrdering;
+            }
         }
 
         /// <summary>

src/core/Akka.Persistence.TCK/Query/CurrentEventsByTagSpec.cs

@@ -157,5 +157,30 @@ public virtual void ReadJournal_query_CurrentEventsByTag_should_find_events_from
             probe2.ExpectNext<EventEnvelope>(p => p.PersistenceId == "b" && p.SequenceNr == 2L && p.Event.Equals("a green leaf"));
             probe2.Cancel();
         }
+
+        [Fact]
+        public virtual void ReadJournal_query_CurrentEventsByTag_should_see_all_150_events()
+        {
+            var queries = ReadJournal as ICurrentEventsByTagQuery;
+            var a = Sys.ActorOf(Query.TestActor.Props("a"));
+
+            for (int i = 0; i < 150; ++i) 
+            {
+                a.Tell("a green apple");
+                ExpectMsg("a green apple-done");
+            }
+
+            var greenSrc = queries.CurrentEventsByTag("green", offset: NoOffset());
+            var probe = greenSrc.RunWith(this.SinkProbe<EventEnvelope>(), Materializer);
+            probe.Request(150);
+            for (int i = 0; i < 150; ++i)
+            {
+                probe.ExpectNext<EventEnvelope>(p => 
+                    p.PersistenceId == "a" && p.SequenceNr == (i + 1) && p.Event.Equals("a green apple"));
+            }
+
+            probe.ExpectComplete();
+            probe.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
+        }
     }
 }