[Persistence] AllEvents Query appears to be missing some events

[SeanKilleen]

Version Information

• Akka.NET: 1.4.34
• Akka.Persistence.Query: 1.4.34
• Akka.Persistence.Query.SQL: 1.4.34
• Akka.Persistence.SqlServer: 1.4.32
• SQL Server: Docker image server:2019-latest via wsl2

Describe the bug

An AllEvents query does not appear to replay all the events.

I have a persistence query to read everything:

readJournal.AllEvents(Offset.NoOffset()).RunForeach(env => writer.Tell(env), materialize);

My persistence-related HOCON:

akka.persistence{
	journal {
	        plugin = "akka.persistence.journal.sql-server"
		sql-server {
			# qualified type name of the SQL Server persistence journal actor
			class = "Akka.Persistence.SqlServer.Journal.SqlServerJournal, Akka.Persistence.SqlServer"

			# dispatcher used to drive journal actor
			plugin-dispatcher = "akka.actor.default-dispatcher"

			# connection string used for database access
			connection-string = "Server=localhost;Database=master;User Id=sa;Password=REDACTED"

			# default SQL commands timeout
			connection-timeout = 30s

			# SQL server schema name to table corresponding with persistent journal
			schema-name = dbo

			# SQL server table corresponding with persistent journal
			table-name = EventJournal

			# should corresponding journal table be initialized automatically
			auto-initialize = on

			# timestamp provider used for generation of journal entries timestamps
			timestamp-provider = "Akka.Persistence.Sql.Common.Journal.DefaultTimestampProvider, Akka.Persistence.Sql.Common"

			# metadata table
			metadata-table-name = Metadata
			
			# Recommended: change default circuit breaker settings
			# By uncommenting below and using Connection Timeout + Command Timeout
			# circuit-breaker.call-timeout=30s
		}
	}

	snapshot-store {
	        plugin = "akka.persistence.snapshot-store.sql-server"
		sql-server {

			# qualified type name of the SQL Server persistence journal actor
			class = "Akka.Persistence.SqlServer.Snapshot.SqlServerSnapshotStore, Akka.Persistence.SqlServer"

			# dispatcher used to drive journal actor
			plugin-dispatcher = ""akka.actor.default-dispatcher""

			# connection string used for database access
			connection-string = "Server=localhost;Database=master;User Id=sa;Password=REDACTED"

			# default SQL commands timeout
			connection-timeout = 30s

			# SQL server schema name to table corresponding with persistent journal
			schema-name = dbo

			# SQL server table corresponding with persistent journal
			table-name = SnapshotStore

			# should corresponding journal table be initialized automatically
			auto-initialize = on
			
			# Recommended: change default circuit breaker settings
			# By uncommenting below and using Connection Timeout + Command Timeout
			# circuit-breaker.call-timeout=30s
		}
	}
	query {
		my-read-journal {
				# Implementation class of the SQL ReadJournalProvider
				class = "Akka.Persistence.Query.Sql.SqlReadJournalProvider, Akka.Persistence.Query.Sql"
  
				# Absolute path to the write journal plugin configuration entry that this 
				# query journal will connect to. 
				# If undefined (or "") it will connect to the default journal as specified by the
				# akka.persistence.journal.plugin property.
				write-plugin = ""
  
				# The SQL write journal is notifying the query side as soon as things
				# are persisted, but for efficiency reasons the query side retrieves the events 
				# in batches that sometimes can be delayed up to the configured `refresh-interval`.
				refresh-interval = 3s
  
				# How many events to fetch in one query (replay) and keep buffered until they
				# are delivered downstreams.
				max-buffer-size = 10000				
		}
	}

Within the actor I have:

            ReceiveAsync<EventEnvelope>(async message =>
            {
                // ....
                capturedEvents++;
                _logger.Warning("RECEIVED EVENT # {EventNumber} in my processing. PersistenceId {PersistenceId} Sequence number {SequenceNumber}", capturedEvents, message.PersistenceId, message.SequenceNr);
                // ....
           }

When running this, I noticed that my logger (logged out to Seq) showed that only a certain event number was processed. When querying the EventJournal table, I noticed there were more events and it contained persistenceIds or some events for those IDs that appeared to be "missing".

Reproduction

I have a private repo that I am willing to invite you to and help configure for your testing if it would be helpful. It will eventually be a public repo.

[SeanKilleen]

I'll update nuget packages to see if it persists with the latest.

CC @Aaronontheweb in case you want to move this to somewhere more relevant.

[SeanKilleen]

I've added a repository with an attempt at a minimal repo at https://github.com/SeanKilleen/2022-04_portfolioss_repro

It contains:

• Database backup and instructions on how to restore it in Docker containers like what I was using
• Minimal console app to replay the events that is taken directly from the code I'm using
• An actor that logs views, and that optionally can attempt to replay them into a DB (if you uncomment the code, run the migrations to create the tables, etc. etc.)

But for now, hopefully the data set and minimal repro will allow you to reproduce reliably.

[Arkatufus]

@SeanKilleen
Thank you for the reproduction repo. I've looked into it and found the problem.
A JSON deserializer exception happened in your reproduction case at around offset 1,200,000, halting the stream. The bug is in our error reporting for the AllEvent stream publisher, reporting the failure at the debug level, and not warning or higher.
We'll work on this ASAP.

[Arkatufus]

@Aaronontheweb we need to formulate a better standard behaviour for Akka.Persistence.Query failure handling and reporting

[SeanKilleen]

Thank you @Arkatufus for the thorough investigation! I'll run with a debug logging level around that offset and see what I turn up to resolve for my use case.

[Aaronontheweb]

@Aaronontheweb we need to formulate a better standard behaviour for Akka.Persistence.Query failure handling and reporting

Probably means having to introduce something either in Akka.Persistence.Query or Akka.Persistence itself in order to enforce this across the board...

[SeanKilleen]

Just to close the loop on my specific issue -- this was due to re-using a database of events across branches.

• In one branch, I was revising my replay actor to track the offset based on some discussions with Aaron. This created a number of snapshots in the snapshot table.
• I then switched back to my main branch, which didn't have the Snapshot class. So it makes perfect sense that Akka would have choked on the attempt to replay the snapshot.
• In my particular case, I could just delete the snapshots from the snapshots table and carry on.

Agreed that surfacing that error during replay as a Warning or higher would have saved me a bunch of time though, so I appreciate you discussing how to achieve it!

[Arkatufus]

@SeanKilleen
After looking closer at the Akka.NET codes, it appears that we did raise the exception inside the stream. It is not signaling any errors in your reproduction code because you're not awaiting the returned Task, effectively making it a detached task. This can be fixed easily by adding a .ContinueWith() at the end to check for stream failure/cancellation/completion.
I've made a PR on your repo to illustrate this: SeanKilleen/2022-04_portfolioss_repro#1