Expose atomicity/transaction boundaries

[makoConstruct]

Summary: Currently, different streaming queries don't know whether they're happening as part of the same DB transaction, or, it seems like there's no way to be notified when a transaction ends. I'm currently trying to build what I think's a pretty good approach to app state persistence, and this is currently in the way. The proposed fix seems simple.

use case:

In my app, and probably in every app I make from now on, I use a kind of persisted signal (pignal) (a lazy observable that persists to storage) for important state. Writes to a signal will be transmitted to the db, and writes to the db will update the signal (the signal subscribes to the db using a streaming select query).

(Also, though this is a minor feature, values are timestamped, which I think may improve behaviour where two isolates are writing at the same time? EG, if one of the isolates sets one of its pignals to 'a' and then sends a write to the db, then the db completes a write from the other isolate of 'b' and so the db tells the pignal that it must now be 'b', and only after that does the pignal's write complete, and then the db tells it it must be 'a' after all. With timestamps it ignores the 'b' write.).

The Signals library supports transactions (batch), wherein multiple signals can be updated without immediately triggering effects, delaying them until the end of the batch, and of course drift supports transactions too, but I can't see a way to get a DB transaction to translate into one Signal transaction.

Right now, a batch of changes in the DB will cause a series of many incomplete/incorrect pignal reactions.

(to get even more specific, the motivating incident was, I'm making a timer app, it has two lists of timers, (pinned timers and timers that're going to be cleaned up automatically if the user leaves them sitting around for too long). The lists are represented by pignals. I'd like the background thread to maintain a third signal that just aggregates both lists as a set, for detecting the creation and deletion of timers. Sometimes the user drags a timer from one list to the other. Currently, when this happens, the aggregate set would erroniously undergo two events (Though I should disclose that for hairy reasons that aren't worth getting into, I don't actually have to fix this in this case. A hack will do. but I could easily imagine that I might have needed this if things had been just a little different!))

proposed solution:

A pair of methods drift.addTransactionBeginListener(()=> signals.startBatch()) and drift.addTransactionEndListener(()=> signals.endBatch()) would do it. These listeners would be invoked before and after every change the DB reports.

[simolus3]

This is a good point, and a weak spot of the stream implementation in drift at the moment (see also: #3338). I have plans for fixing this properly, but it requires fairly large architectural changes to the package.

Inside a transaction, drift is already batching detected writes to emit them atomatically. That's the first step of what you want, but then when streams are invalidated they all run independently.

I'm not sure an addTransactionBeginListener / addTransactionEndListener would help here. Is the idea that you would throw results away if we started fetching them before a transaction boundary?

I think a good API would allow "bundling" multiple stream queries so that they always emit a consistent snapshot. That's the suggestion from the other issue, but that's challenging to implement on what drift has today. We could lock the entire database while fetching those streams, but that's also inefficient.
Really, we should have "read-only transactions". With WAL mode, these can be active as long as they need to be without blocking a writer or other readers. And within a single read transaction, we can issue many reads that are guaranteed to all see the exact same data. So that would be an excellent fit for this, but adding new functionality that affects the executor is a breaking change.

[makoConstruct]

Throw results away? No? The idea with addTransactionBeginListener / addTransactionEndListener is that transaction start listeners are always invoked before any of the streams receive updates, and the transaction end listeners wont be invoked until all of the streaming queries affected by that transaction have been invoked, and will be invoked before the next transaction starts.

Are there db events that technically aren't part of a transaction? If so they'd each be treated as being very small transaction here. Or, maybe if an event is atomic there's no need to invoke the listeners (that would be fine for my usecase).

[simolus3]

and the transaction end listeners wont be invoked until all of the streaming queries affected by that transaction have been invoked

I see - but even that is hard to implement with the current implementation because there's no "wait to update stream queries" state we could await. Every stream is an independent async system and there's no way to await a refresh of streams on the database.

Would the API suggested in #3338 work for you? Since this will require some internal restructuring anyway, I want to get the APIs right this time :D A transaction listener that implicitly serializes all statements running on the drift database seems a bit fragile and not in spirit of WAL mode where a reader can operate without blocking a writer.

Right now, a batch of changes in the DB will cause a series of many incomplete/incorrect pignal reactions.

This is the fundamental issue worth fixing here, right? It sounds to me that if you're using a stream group as suggested in that issue, this would no longer be an issue.

[makoConstruct]

That API seems less than ideal. It would work for the specific example given, but I guess one of the insights of reactive programming is that the data dependency relationships can change dynamically over time, it wouldn't be in that spirit, since this approach requires declaring the synchronization between these pignals when they're created, and new pignals can't be added to that synchronization group later (right?).

Does Drift (or sqlite) know which streaming queries are entangled by a transaction? (I'm guessing on some level the system knows that, and I want to frame the problem as that information getting lost somewhere along the way)

I don't understand why "serialising" could slow things down. Pignal transactions never span over multiple isolates, the pignals are all in the same isolate, they don't run in parallel.

[simolus3]

but I guess one of the insights of reactive programming is that the data dependency relationships can change dynamically over time

Good point. I still think a transaction listener / hook API is a bit to low-level, but there should be alternatives. We could probably allow a synchronization group to grow / shrink dynamically, but I'm also thinking that this should probably be the default for all streams. In a WAL setup with a pool of readers, we wouldn't really be blocking anything by running something in a transaction.

So I imagine something that could work for you without requiring new drift APIs is:

1. You listen on table changes manually.
2. When a table changes, start a transaction with a to-be-added readonly mode and call signals.startBatch().
3. Update affected pignals.
4. Call signals.endBatch() and complete the read transaction.

[makoConstruct]

Sounds fine. Disposition: close this issue when readonly is in (why do I need readonly btw?).

Would this make it harder to have situations where we have lots of isolates and they're all listening to small, nonoverlapping parts of the database? Would they all burn cpu every time any part of the pignals table changes?

[simolus3]

why do I need readonly btw

Semantically, this would already be possible today with a regular transaction. However, we can only have one transaction at a time, so running all the readers would block writes to the database while we're fetching results. That's only a minor issue in practice since drift doesn't even set up multiple connections by default, but we really want one write connection and an arbitrarily-sized pool of readers can operate in parallel.

where we have lots of isolates and they're all listening to small, nonoverlapping parts of the database? Would they all burn cpu every time any part of the pignals table changes?

In the sense that they'd re-run queries? Yes, but that's also how drift streams work today. There's a per-table granularity on how updates propagate, so if you have a single pignals table to store results, all your pignals would be read again.

[makoConstruct]

Wait, so, (I'm just debugging an issue where rapid successive writes are being ignored, I think it's probably an issue on my end. But the output confirmed and reminded me of the following concern) every streaming query that shares a table with any value that was changed will receive an update even if its own value hasn't changed?.. all of my pignals are in one table, so this would imply cpu use that's quadratic on the size or complexity of the app :(

[simolus3]

every streaming query that shares a table with any value that was changed will receive an update even if its own value hasn't changed?

Correct - drift would re-run all those queries on a per-table granularity. If you want to filter duplicates early, you can use distinct() on the stream. Obviously you'd still pay the cost of running a bunch of queries more often than necessary. Drift doesn't really have a good solution for that at the moment, you'd have to implement your own tracker if you want rowid-level filters.