so ...done a bit of experimenting (not a lot), and as far as I can see, stereamstone doesn't solve any problems I have. Originally I thought it would help solve the problem of ...what happens if azure decides to create a second instance of an azure function without me expecting it to, despite any attribute or json or configurations requiring a single partition or any other trick to try to force an azure function into some type of singleton
private static Engine<AccountModel> _engine;
// static constructor for Azure function class Test1
static Test1()
{
// safety net in case not stopped.
AppDomain.CurrentDomain.ProcessExit += (s, e) => {
if(!stopped)
{
_engine.DisposeAsync().GetAwaiter().GetResult();
}
};
}
Hi @rofr. Recently I was looking into ways to hold a large singleton aggregate in memory but still have reasonable persistence to DB/file/etc. in order to recover state. I eventually got into event sourcing, Prevayler, and finally, memstate.
When I look at the github page, there appears to be infrequent development, and it's still in alpha/beta stage. Could you give some idea on how stable the current release is and what your intentions are for further development/support?
In addition, I have a question about using Postgres as the persistence. Suppose in the write-ahead approach, memstate attempts to persist a command C1 to Postgres. For whatever reason, the server persists C1, but we experience a client-side timeout. At this point, the in-memory state will be behind that of the persistence. If we then receive a command C2, that will persist to Postgres and then operate on the stale in-memory state and possibly result in a different state than when the commands are later replayed. Is there some mechanism in memstate to detect this kind of out-of-sync behavior (e.g. using event versioning) and recover transparently?
Hi @albertyfwu
I'll take the easy question first..
In addition, I have a question about using Postgres as the persistence. Suppose in the write-ahead approach, memstate attempts to persist a command C1 to Postgres. For whatever reason, the server persists C1, but we experience a client-side timeout. At this point, the in-memory state will be behind that of the persistence. If we then receive a command C2, that will persist to Postgres and then operate on the stale in-memory state and possibly result in a different state than when the commands are later replayed. Is there some mechanism in memstate to detect this kind of out-of-sync behavior (e.g. using event versioning) and recover transparently?
JournalRecords have a RecordNumber and by default these are required to have an unbroken sequence. In the scenario above, the engine will throw an exception when receiving C2 if it has not yet seen C1.
When I look at the github page, there appears to be infrequent development, and it's still in alpha/beta stage. Could you give some idea on how stable the current release is and what your intentions are for further development/support?
We are running memstate in production for several systems using EventStore or SqlStreamStore for storage. The core features are solid using either of these storage providers. There are some loose ends that need to be addressed before a 1.0 release though. Event subscriptions over a remote connection are not working for example.
PS: don't use the standalone Postgres provider, we may drop it altogether in favor of SqlStreamStore which has support for MySql, MSSQL and Postgres.
Hi @albertyfwu
I'll take the easy question first..In addition, I have a question about using Postgres as the persistence. Suppose in the write-ahead approach, memstate attempts to persist a command C1 to Postgres. For whatever reason, the server persists C1, but we experience a client-side timeout. At this point, the in-memory state will be behind that of the persistence. If we then receive a command C2, that will persist to Postgres and then operate on the stale in-memory state and possibly result in a different state than when the commands are later replayed. Is there some mechanism in memstate to detect this kind of out-of-sync behavior (e.g. using event versioning) and recover transparently?
JournalRecords have a RecordNumber and by default these are required to have an unbroken sequence. In the scenario above, the engine will throw an exception when receiving C2 if it has not yet seen C1.
Thanks for the quick response. I'm still unclear on when this discrepancy is resolved. After C1 is written to the DB (but unknown to the engine), how does the engine know to throw an exception when receiving C2? At that time, it wouldn't have learned that C1 was successfully written to DB yet. Is there some syncing that's happening out-of-band between C1 and C2? Or is there some synchronous synchronization happening to resolve discrepancies between engine in-memory and DB at the time it persists C2 to DB? Let me know if I need to explain this question more clearly.
In my particular use case, I have one single big aggregate (instead of many instances).
I know that memstate persists/applies commands sequentially, but I actually need the aggregate "locked" before the "commands" are issued, since I'm doing something more like "event" rather than "command" sourcing.
My workflow is:
- Lock the single big aggregate.
- Validate the command.
- If success, create an event, persist it, and apply it.
- Unlock the single big aggregate.
It seems that memstate supports something like this instead:
- Validate the request.
- If success, create a command.
- Lock the aggregate.
- Persist the command and apply it.
- Unlock the aggregate.
To me, it seems that I'd need to handle my own locking (before validation) before handing it off to memstate. Let me know if see how this can be more natively handled by memstate. Thanks.
After C1 is written to the DB (but unknown to the engine), how does the engine know to throw an exception when receiving C2?
Because C2 has the wrong RecordNumber. How it works is not trivial to explain as the various storage providers behave differently. An RDBMS backend could use either auto_increment or optimistic concurrency to ensure that commands are assigned sequential ids.
- You create a command object and pass it to
engine.Execute(cmd) - The engine sends the command to the underlying storage provider
- Storage provider writes the command
- The command as applied to the aggregate
- Validate your request with a memstate Query and issue a Command if the validation was successful
- Validate inside the Command.Execute method before making any changes to the aggregate. If invalid throw an exception or return a status code
After C1 is written to the DB (but unknown to the engine), how does the engine know to throw an exception when receiving C2?
Because C2 has the wrong RecordNumber. How it works is not trivial to explain as the various storage providers behave differently. An RDBMS backend could use either auto_increment or optimistic concurrency to ensure that commands are assigned sequential ids.
@rofr Thanks for all your help so far, but I don't think we're talking about the same thing.
I do agree that using auto_increment or other techniques can guarantee that commands are assigned sequential ids in the DB. However, I don't see how that addresses keeping the in-memory representation in-sync with the DB in a certain client-side timeout (note: not server-side timeout) situation:
- Suppose we start with the state
s0for our in-memory model. - We call
engine.Execute(c1), which sends command to DB. - DB writes a new row
c1with ID1(auto-incrementing). engine.Execute(c1)gets a client-side timeout because the DB write took a long time.- At this point, the
engineis unaware that rowc1was created in the DB, and in-memory model stays as states0. - At this point, if we call
engine.Execute(c2), that will write the command to DB (rowc2with ID2), and the in-memory model will try to computec2(s_0). - However, if at this point the application crashes, and we replay events from the DB, we'd try to replay as
c2(c1(s_0))instead.
How does memstate either (a) prevent c2 from being written to DB (and then immediately re-sync in-memory with DB), or (b) avoid being out-of-sync in the first place?
Either I am misunderstanding what the code is actually doing, or there are some additional synchronization steps above that I'm unaware of.
Thanks again. I appreciate it.
INSERT c2 if current max_seq_id is <expected_seq_id_from_in_memory_model> (or we don't auto-increment in DB but use the in-memory seq_id to write, and enforce uniqueness on the seq_id column in DB).
I actually spent a couple hours reading the code and I think I have the misunderstanding.
I originally thought that the in-memory model was immediately updated following a successful command write to DB, but I now see that memstate was written to support horizontally-scaled services in a traditional CQRS manner, where in-memory state isn't updated until the subscriber reads the command from the DB, which guarantees correctness.
Given that my particular use case runs in a single process (no horizontal scaling, ever), I'm wondering if that simplification allows me a way for me to achieve the write-ahead logging semantic but get immediate read-after-write instead of waiting for subscribing to my own writes to DB. But in any case, that doesn't seem to be the use case for memstate.
Thanks.
For 1 you would need to ensure that no other Command is executed between your Query and Command which is impossible in a load balanced environment
That doesn't seem sufficient; suppose:
engine.execute(C1), which gets persisted to DB.- I lock the aggregate, and issue a Query followed by
engine.execute(C2)(before subscription picks up onC1having been written to DB). - Now DB has both
C1andC2, thoughC2was persisted assuming that there was no state change between Query (which really reflected state beforeC1was applied) andengine.execute(C2).
I originally thought that the in-memory model was immediately updated following a successful command write to DB, but I now see that memstate was written to support horizontally-scaled services in a traditional CQRS manner, where in-memory state isn't updated until the subscriber reads the command from the DB, which guarantees correctness.
Not really CQRS as there is only a single model for both reads and writes. But yes, memstate is designed to support multiple nodes, each with the same sequence of in-memory states.
@albertyfwu the docs are copied from origodb, and snapshots are not yet implemented. The main reason being we tend to avoid them and just replay the entire journal at startup. A huge and complex snapshot can take just as long or longer to deserialize than simply rerunning the commands. @goblinfactory did some experiments using Azure table storage and saw 75K commands/second replay rate. I have systems with millions of commands that take ~30 minutes to load. Not saying snapshots aren't useful just not as necessary as some may think.
my test was on a Virtual machine running on my laptop, and with JSON serialisation, with no attempts at optimising for speed, e.g. using wire or protobuf. I'm on leave in 2 weeks and will redo those tests so that I can update the docs with some real numbers.
I noticed in the docs that you mention snapshots, but I couldn't find any corresponding code actually supporting that. Is this a future feature? https://memstate.io/docs/core-1.0/storage/snapshots/
for my use case I will probably look at compressing my journal records into batches as a background or scheduled service depending on the usage patterns to reduce my storage costs with azure. currently my poc is stupidly simple where I write a new journal entry per command written each time to azure table storage. however I can "optimise" the storage after the fact, i.e take every n records and convert that into a batch and write batches to blob storage, ..then when reading I take a pointer to the top of the stream of journals and batches.
engine.Execute(Command) returns a Task which completes when the command has been applied to the in-memory state. If you have a single engine and do reads and writes (awaited) on a single thread then the system is 100% ACID.
It seems that this solution still has one potential edge case (client-side timeout):
Suppose we have a function like this:
async Task RunStuff(Command cmd)
{
await _semaphoreSlim.WaitAsync();
try
{
var state = await _engine.Execute(_query);
// Do some validation of `cmd` against `state`.
// It's valid, so we proceed.
await _engine.Execute(cmd);
}
finally
{
_semaphoreSlim.Release();
}
}- Model is at state s0 (and so is DB).
- We call
RunStuff(c1), which gets state s0, validates c1, and runsengine.Execute(c1). engine.Execute(c1)succeeds in persisting to DB, but we experience client-side timeout and think it failed.RunStuff(c1)returns and unlocks.- Somebody else calls
RunStuff(c2), which could still read state s0, as engine may not yet have synced with database to see newc1written. RunStuff(c2)reads state s0, validates c2, and writes it to DB.- Now, c1 and c2 are persisted in DB and the in-memory model will be synced soon thereafter.
The issue now is that c2 was validated assuming the previous state was s0 instead of s1 = c1(s0).
It seems to me the only ways to resolve this kind of issue (in the case of a single process, and not load-balanced) are:
- Have the
engineitself write its expected journal record numbers to DB (soc2using the same record number asc1would get rejected by column uniqueness constraint in DB). - Expose a public API on
enginethat allows waiting on the next DB sync job/thread to complete successfully. This way we can prevent the secondRunStuff(c2)call from even being able to acquire the semaphore lock until we resolve that DB sync job.
The first proposal obviously doesn't work for load-balanced architectures, but the second proposal seems non-breaking. Do you see a better way? @rofr
Suppose we have a function like this:
You are silently dropping an exception, don't do that.
Expose a public API on engine that allows waiting on the next DB sync job/thread to complete successfully
That is exactly what await engine.Execute(cmd); does in your code example. It won't timeout, it will wait forever. But we haven't really explored these system failure scenarios so you're raising some really good questions here. System tests for failure conditions would be really good to have and will certainly uncover some bugs. At the least, behavior needs to be well defined and documented.
@rofr @goblinfactory Do you 2 have any recommendations for an easy way for me to incorporate snapshotting while using memstate?
I noticed the following two lines in Engine.OnRecordReceived(), showing that while the Kernel locks reads/writes for command/query execution, that lock doesn't apply to _lastRecordNumber:
var result = _kernel.Execute(record.Command);
Interlocked.Exchange(ref _lastRecordNumber, record.RecordNumber);However, to store a proper snapshot, I'd need to atomically get both the current state AND the last record number.
Solution 1:
If I have a single process writing to memstate, I can just sequentialize all writes to Engine, await them, and write snapshots. This works, but isn't native.
Solution 2:
I fork memstate and move the read/write lock out of Kernel and into Engine, so that I can ensure that reads to my model for snapshotting can get an atomic pair of (snapshot, last record number).
Is there an easier solution you see that I'm overlooking? How did you implement this in origodb?
Thanks.
Hi @rofr I trying to create a generic database helper that creates a memstate database for me, as well as automatically creates a membership database to live next to the database<T>, in order to do that I need to be able to read/write to two instances of memstate at the same time. I created the following experiment to test (very roughly) if this is possible, ...
is the following safe to do ? i.e. create two instances, but setting the config for first, create first db, then ...call Config.Current and change path, and create second wit the same configuration. It seems to work?
is this safe to do at scale? -> https://gist.github.com/goblinfactory/de0752827621dac47ea82d637d1a712a - i.e. access two memstate instances that appear to somewhat share some configuration.
The obvious problem with this strategy that I've already encountered is that there's way to create a Command<TModel, TResult> and in that command, write to two databases at the same time. I was also unable to create a generic database, that I can reference as a package that can allow me to create a FooDb : SomeBaseDb where some base Database comes with some pre-setup behaviors with the way that Command<TModel, TResult> works. The moment you create a base database class, you can no longer create commands from anything derived from that database class, ..or at least i was unable to find a way to get around the generic restrictions.