Assess performance and integrity of Redis data storage + serialization

[akkadotnet]

Assess performance and integrity of Redis data storage + serialization

[Aaronontheweb]

Looks like we're using bespoke serialization for Akka.Persistence.Redis: #33

Same issue happened with almost all of the Akka.Persistence plugins: akkadotnet/akka.net#3811

edit: this is only for snapshots - not sure how the journal does it just yet...

[Aaronontheweb]

Looks like we might be doing the right thing for journal serialization:

Akka.Persistence.Redis/src/Akka.Persistence.Redis/JournalHelper.cs

Lines 23 to 33 in 392f9b8

	public byte[] PersistentToBytes(IPersistentRepresentation message)
	{
	var serializer = _system.Serialization.FindSerializerForType(typeof(IPersistentRepresentation));
	return serializer.ToBinary(message);
	}
	public IPersistentRepresentation PersistentFromBytes(byte[] bytes)
	{
	var serializer = _system.Serialization.FindSerializerForType(typeof(IPersistentRepresentation));
	return serializer.FromBinary<IPersistentRepresentation>(bytes);
	}

Compare that to what we do in MongoDb:

https://github.com/akkadotnet/Akka.Persistence.MongoDB/blob/481fa679199da704ab9eed4d0a504818363a1dcc/src/Akka.Persistence.MongoDb/Journal/MongoDbJournal.cs#L386-L396

[Aaronontheweb]

The actual Redis data structure layouts, at first glance, seem sensible - but the low performance numbers might be due to the use of the Redis channels for firing events that drive Akka.Persistence.Query:

Akka.Persistence.Redis/src/Akka.Persistence.Redis/Journal/RedisJournal.cs

Lines 90 to 130 in 392f9b8

	var transaction = Database.CreateTransaction();
	var payloads = aw.Payload.AsInstanceOf<IImmutableList<IPersistentRepresentation>>();
	foreach (var payload in payloads)
	{
	var (bytes, tags) = Extract(payload);
	// save the payload
	transaction.SortedSetAddAsync(_journalHelper.GetJournalKey(payload.PersistenceId), bytes, payload.SequenceNr);
	// notify about a new event being appended for this persistence id
	transaction.PublishAsync(_journalHelper.GetJournalChannel(payload.PersistenceId), payload.SequenceNr);
	//save events sequenceNr and persistenceId so that we can read all events
	//with it starting from a given sequenceNr
	var journalEventIdentifier = $"{payload.SequenceNr}:{payload.PersistenceId}";
	transaction.ListRightPushAsync(_journalHelper.GetEventsKey(), journalEventIdentifier);
	// notify about this event
	transaction.PublishAsync(_journalHelper.GetEventsChannel(), journalEventIdentifier);
	// save tags
	foreach (var tag in tags)
	{
	transaction.ListRightPushAsync(_journalHelper.GetTagKey(tag), journalEventIdentifier);
	transaction.PublishAsync(_journalHelper.GetTagsChannel(), tag);
	}
	}
	// set highest sequence number key
	transaction.StringSetAsync(_journalHelper.GetHighestSequenceNrKey(aw.PersistenceId), aw.HighestSequenceNr);
	// add persistenceId
	transaction.SetAddAsync(_journalHelper.GetIdentifiersKey(), aw.PersistenceId).ContinueWith(task =>
	{
	if (task.Result)
	{
	// notify about a new persistenceId
	Database.Publish(_journalHelper.GetIdentifiersChannel(), aw.PersistenceId);
	}
	});

We might want to add the option to disable those for users who aren't using Akka.Persistence.Query with this plugin.

[Aaronontheweb]

Validated that Akka.Persistence.Query has been the source of some significant performance problems and distribution problems with Redis cluster: #114

Validated that Akka.Persistence.Redis is following Akka.NET best practices for journal serialization: #117