Re-implement PersistenceIds persitence query to match scala implementation behavior #4531

Arkatufus · 2020-07-29T18:10:51Z

Related to #4524

The current implementation PersistenceIds query does not pull its updates from the database, instead, it relies on an internal cache that only caches messages that passes the ActorSystem locally. This became a problem on a multi-ActorSystem system, as it does not capture the state of the whole system.

Changes:

Subscription and publishing are now done using FanoutPublisher.
FanoutPublisher buffer/cache is expanded to handle long records size.
Add a new type of FanoutPublisher buffer/cache that can handle distinct and unbounded records cache.
FanoutPublisher is dynamically created and destroyed based on subscriber count.

…rover list.

…ocessorImpl

Aaronontheweb

Couldn't finish the entire review, but wanted to submit what I have so far

Aaronontheweb · 2020-08-05T15:20:19Z

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

@@ -141,7 +141,7 @@ internal sealed class LiveAllEventsPublisher : AbstractAllEventsPublisher
        public LiveAllEventsPublisher(long fromOffset, TimeSpan refreshInterval, int maxBufferSize, string writeJournalPluginId)
            : base(fromOffset, maxBufferSize, writeJournalPluginId)
        {
-            _tickCancelable = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(refreshInterval, refreshInterval, Self, EventsByTagPublisher.Continue.Instance, Self);
+            _tickCancelable = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(refreshInterval, refreshInterval, Self, AllEventsPublisher.Continue.Instance, Self);


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

Aaronontheweb · 2020-08-05T17:46:21Z

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

+
+            return Source.FromPublisher(_persistenceIdsPublisher)
+                .MapMaterializedValue(_ => NotUsed.Instance)
+                .Named("AllPersistenceIds") as Source<string, NotUsed>;


Just do a direct cast here rather than an as - that way we get a hard failure instead of passing null back to the consumer.

The cast was not necessary anyway, I think it's there as a readability helper. I'll remove it.

Aaronontheweb · 2020-08-05T17:47:24Z

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

-            .Named("AllPersistenceIds") as Source<string, NotUsed>;
+        public Source<string, NotUsed> PersistenceIds()
+        {
+            if (_persistenceIdsPublisher is null)


This call might need to be synchronized - we want all of the consumer of the PersistentIds query to receive the same instance, right?

Can you check and see if my implementation looks good?

Aaronontheweb · 2020-08-05T17:47:34Z

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

+
+        private void PersistenceIdsShutdownCallback()
+        {
+            _persistenceIdsPublisher = null;


Same comment as above, RE: synchronization

Aaronontheweb · 2020-08-05T17:50:21Z

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

@@ -326,7 +331,8 @@ protected AbstractQueryExecutor(QueryConfiguration configuration, Akka.Serializa

            AllPersistenceIdsSql = $@"
                SELECT DISTINCT e.{Configuration.PersistenceIdColumnName} as PersistenceId 
-                FROM {Configuration.FullJournalTableName} e;";
+                FROM {Configuration.FullJournalTableName} e
+                WHERE e.{Configuration.OrderingColumnName} > @Ordering";


So to get all persistent Ids we query from the journal? What happens in the event that a persistent Id exists but only has a snapshot currently? This happens with AtLeastOnceDelivery actors frequently.

The Metadata table will always store the full set of persistent Ids - that's where we retrieve the maximum sequence number from for each entity, and I don't believe we allow any persistent entities to have their information deleted from the metadata table either.

Ah, I didn't know that. So I need to do a join of the same query against the Journal and Metadata table?

Just query Metadata directly - no need to touch the journal.

This implementation will be different for each plugin - we should add a TCK method to check to see that that PersistenceIds query can return a PersistenceId for an entity that saves only snapshots. Could you add that?

Done, reference removed.

Done, added test that checks for PersistenceIds from both snapshotted and journaled actors.

src/core/Akka.API.Tests/CoreAPISpec.ApproveStreams.approved.txt

Aaronontheweb · 2020-08-05T18:51:39Z

src/core/Akka.API.Tests/CoreAPISpec.ApproveStreams.approved.txt

@@ -3016,6 +3024,17 @@ namespace Akka.Streams.Implementation
        void WaitForUpstream(int waitForUpstream);
    }
    public interface ISpecViolation { }
+    public interface IStreamBuffer<T>


Did you come up with this or is this a port from the JVM?

I came up with it on my own, I have to have an interface to base my custom buffer.

Want to make this internal? Any reason it needs to be public?

Yeah, I'll move it to internal, user should never touch that

Done, moved to internal.

Aaronontheweb · 2020-08-05T18:54:18Z

src/core/Akka.Persistence.TCK/Akka.Persistence.TCK.csproj

@@ -10,6 +10,7 @@
  </PropertyGroup>

  <ItemGroup>
+    <ProjectReference Include="..\..\contrib\persistence\Akka.Persistence.Query.Sql\Akka.Persistence.Query.Sql.csproj" />


Why does the Query.Sql library need to be included inside the TCK for all persistence plugins generally, including non-SQL ones?

I need to access some of its internal class in the test so I can write a test to check that the buffer is actually released when it ran out of subscribers.

The TCK should work broadly, without any connection to a specific plugin, since the TCK is what's used to validate all plugins. We'll need to clean that up.

Got it, I'll move it to Query.Sql.Testkit instead.

Done, moved to PersistenceIdsSpec instead

…t for snapshot and journal case.

…to Fix_PersistenceIds

Aaronontheweb

Some more changes needed to pagination for PersistenceIds

Aaronontheweb · 2020-08-18T12:41:59Z

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

+                FROM (
+                    SELECT DISTINCT e.{Configuration.PersistenceIdColumnName} as Id 
+                    FROM {Configuration.FullJournalTableName} e
+                    WHERE e.{Configuration.SequenceNrColumnName} > @SequenceNr


I don't think we should try to add pagination support for this query - the sequence number is unique to each entity and thus can't be used for ordering. I'd just do a full scan of the index for the time being since that's a covered query.

If we start getting complaints about non-performance, we can look at adding a "date added" timestamp that indicates the age when this persistent Id was first used and use that as the Ordering for a query.

ROWID won't work for SQL Server and other RDBMS because it's not consistent across multiple queries on the same table: https://stackoverflow.com/a/58916093/377476

Aaronontheweb · 2020-08-18T12:42:16Z

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

        /// <returns>TBD</returns>
-        Task<ImmutableArray<string>> SelectAllPersistenceIdsAsync(DbConnection connection, CancellationToken cancellationToken);
+        Task<ImmutableArray<string>> SelectAllPersistenceIdsAsync(DbConnection connection, CancellationToken cancellationToken, long offset);


I'd skip pagination

Aaronontheweb · 2020-08-18T12:42:49Z

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

+        {
+            get
+            {
+                _lock.EnterReadLock();


Stick with just a simple lock keyword here - no need for a ReaderWriterLock. This query likely won't be subject to large amounts of contention - better to go with something simple.

Aaronontheweb · 2020-08-18T12:43:43Z

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

@@ -77,6 +88,36 @@ public virtual void ReadJournal_AllPersistenceIds_should_find_new_events_after_d
            });
        }

+        [Fact]
+        public virtual void ReadJournal_AllPersistenceIds_should_find_events_on_both_journal_and_snapshot_store()


This is excellent

Aaronontheweb · 2020-08-18T13:11:21Z

src/core/Akka.Streams/Implementation/ResizableMultiReaderRingBuffer.cs

@@ -72,27 +199,27 @@ public interface ICursor
    /// </summary>
    /// <typeparam name="T">TBD</typeparam>
    [InternalApi]
-    public class ResizableMultiReaderRingBuffer<T>
+    public class ResizableMultiReaderRingBuffer<T> : IStreamBuffer<T>


Aaronontheweb · 2020-08-18T13:16:07Z

src/core/Akka.Streams/Implementation/SubscriberManagement.cs

                if (!_buffer.Value.Write(value))
                    throw new IllegalStateException("Output buffer overflow");
-                if (Dispatch(_subscriptions))
+                if (_buffer.Value.Length > oldBufferLength && Dispatch(_subscriptions))


How does this work exactly? Wouldn't we just want to pull so long as there's demand?

Good catch, that's not supposed to be there. I'm fixing it right now.

Aaronontheweb · 2020-08-18T13:19:56Z

src/core/Akka.Streams/Util/BitOperations.cs

+// http://graphics.stanford.edu/~seander/bithacks.html
+namespace Akka.Streams.Util
+{
+    // TODO: replace this with the official System.Numerics.BitOperations when we move on to .NET Core 3.0


…ormance reasons

src/core/Akka.Streams/Implementation/SubscriberManagement.cs

Aaronontheweb · 2020-08-19T22:34:02Z

Akka.Persistence.Sqlite.Tests.Batching.BatchingCurrentSqliteAllEventsSpec.ReadJournal_query_AllEvents_should_complete_when_no_events
Failed: Expected a message of type Akka.Streams.TestKit.TestSubscriber+OnComplete, but received {TestSubscriber.OnNext(EventEnvelope(persistenceId:a, seqNr:1, offset:Akka.Persistence.Query.Sequence, event:hello))} (type Akka.Streams.TestKit.TestSubscriber+OnNext`1[Akka.Persistence.Query.EventEnvelope]) instead  from [akka://BatchingCurrentSqliteAllEventsSpec/user/StreamSupervisor-14/Flow-0-0-actorPublisherSource-CurrentAllEvents-actorPublisherSource#930891650]
Expected: True
Actual:   False

And

Akka.Persistence.Sqlite.Tests.Batching.BatchingSqlitePersistenceIdSpec.ReadJournal_AllPersistenceIds_should_find_new_events_after_demand_request
System.Exception : Expected OnNext(_), yet no element signaled during 00:00:10

Both failed in an earlier run in this PR - any idea why?

Arkatufus · 2020-08-19T22:35:19Z

No idea, I'd have to look more into that

Aaronontheweb · 2020-08-19T22:37:03Z

Let's check it out so we don't have any more racy tests in the suite -

…t used twice

Aaronontheweb · 2020-08-20T18:26:21Z

src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Batching/BatchingSqliteAllEventsSpec.cs

@@ -23,13 +21,13 @@ class = ""Akka.Persistence.Sqlite.Journal.BatchingSqliteJournal, Akka.Persistenc
                table-name = event_journal
                metadata-table-name = journal_metadata
                auto-initialize = on
-                connection-string = ""Filename=file:memdb-journal-eventsbytag-{id}.db;Mode=Memory;Cache=Shared""
+                connection-string = ""Filename=file:memdb-journal-eventsbytag-{Guid.NewGuid()}.db;Mode=Memory;Cache=Shared""


CumpsD · 2020-09-02T20:39:23Z

@Aaronontheweb don't forget to implent this here: https://github.com/AkkaNetContrib/Akka.Persistence.PostgreSql/blob/dev/src/Akka.Persistence.PostgreSql/Journal/PostgreSqlQueryExecutor.cs

Aaronontheweb · 2020-09-02T22:04:21Z

@akkadotnet/core - should we move the PostgreSql and MySQL plugins out of AkkaNetContrib and into the main organization?

CumpsD · 2020-09-03T06:33:14Z

Would be nice to see them as first class providers :)

Arkatufus added 4 commits July 29, 2020 22:04

Rebase to new dev

3a6af76

Re-implement PersistenceIds Query

ef43856

Fix copy-paste error

a7e2e5f

Make sure APIs are as backward compatible as possible, update API app…

7796f2f

…rover list.

Arkatufus requested a review from Aaronontheweb July 29, 2020 18:33

Aaronontheweb mentioned this pull request Jul 30, 2020

Akka.Persistence.Sql SqlJournal caching all Persistence Ids in memory does not scale #4524

Closed

Arkatufus added 5 commits July 31, 2020 05:37

Merge branch 'dev' into Fix_PersistenceIds

ea97555

Add unit test for publisher (cache) deallocation.

f98a9d7

Move onTerminate callback from AllPersistenceIdsPublisher to FanoutPr…

edb8e52

…ocessorImpl

Update API approver list

238fca8

Merge branch 'dev' into Fix_PersistenceIds

f8c8460

Aaronontheweb requested changes Aug 5, 2020

View reviewed changes

Arkatufus added 9 commits August 8, 2020 04:59

Merge branch 'dev' into Fix_PersistenceIds

1d8f659

Remove redundant as cast

738f855

Remove reference to Akka.Persistence.Query.Sql from Akka.Persistence.TCK

1b79bf5

Add concurrency locking on the internal persistence ids publisher.

1352b0e

Make PersistenceIds to query both Journal and Metadata table. Add tes…

1ec20e4

…t for snapshot and journal case.

Change IStreamBuffer access to internal.

e983d47

Merge branch 'dev' into Fix_PersistenceIds

a7c5f81

Update API approver list

f6f1b60

Merge branch 'Fix_PersistenceIds' of github.com:Arkatufus/akka.net in…

9e70477

…to Fix_PersistenceIds

Aaronontheweb requested changes Aug 18, 2020

View reviewed changes

Arkatufus added 6 commits August 18, 2020 23:55

Go back to using ordeting column to filter the journal table for perf…

fabeaa1

…ormance reasons

Change synchronization to a simpler lock

386d6fc

Change test to reflect changes in synchonization

e7d174a

Update persistence ids SQL in BatchingSqlJournal

6982944

Change how SubscribeManager checks for available data.

5a08ec5

Merge branch 'dev' into Fix_PersistenceIds

8df7e67

Arkatufus commented Aug 19, 2020

View reviewed changes

src/core/Akka.Streams/Implementation/SubscriberManagement.cs Outdated Show resolved Hide resolved

Merge branch 'dev' into Fix_PersistenceIds

4eb24e2

Aaronontheweb approved these changes Aug 19, 2020

View reviewed changes

Arkatufus added 2 commits August 20, 2020 01:50

Change confusing Count() to AvailableData property

e5f7d98

Update API approver list

8954464

Arkatufus added 3 commits August 20, 2020 23:38

Make sure that tests are never run in parallel

c20ddb3

Prune excessive test

56f4f47

Append a GUID on the db filename to make doubly sure that db never ge…

d75c717

…t used twice

Aaronontheweb reviewed Aug 20, 2020

View reviewed changes

Aaronontheweb merged commit d612c89 into akkadotnet:dev Aug 20, 2020

This was referenced Aug 20, 2020

added v1.4.10 release notes #4546

Merged

v1.4.10 Stable Release #4547

Merged

Arkatufus deleted the Fix_PersistenceIds branch March 8, 2021 13:31

Re-implement PersistenceIds persitence query to match scala implementation behavior #4531

Re-implement PersistenceIds persitence query to match scala implementation behavior #4531

Conversation

Arkatufus commented Jul 29, 2020

Aaronontheweb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Arkatufus Aug 17, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Aaronontheweb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Aaronontheweb commented Aug 19, 2020

Arkatufus commented Aug 19, 2020

Aaronontheweb commented Aug 19, 2020

Choose a reason for hiding this comment

CumpsD commented Sep 2, 2020 • edited

Aaronontheweb commented Sep 2, 2020

CumpsD commented Sep 3, 2020

Arkatufus Aug 17, 2020 •

edited

CumpsD commented Sep 2, 2020 •

edited