Require both KVStore and KVStoreSync implementations, switch BP to be fully-async

[tnull]

As an intermediary step towards making our IO fully async, we now require any store to implement both KVStore and KVStoreSync, which allows us to switch over to the fully-async background processor and take further migration steps bit-by-bit when we make more and more of the core codebase async.

To this end, we refactor VssStore and SqliteStore to implement KVStore

TODOs:

• Implement KVStore for TestStore upstream to fix tests
• Implement write-order tracking for the VSS KVStore implementation

.. draft until then.

[ldk-reviews-bot]

👋 Thanks for assigning @joostjager as a reviewer!
I'll wait for their review and will help manage the review process.
Once they submit their review, I'll check if a second reviewer would be helpful.

[tnull]

This now builds based on the just-merged lightningdevkit/rust-lightning#4069. We have yet to add write-version tracking for VSS.

[tnull]

Should be good for review. For the VssStore versioning I copied as much as possible from the already-reviewed approach over at lightningdevkit/rust-lightning#3931

[tnull]

Rebased to address conflicts post-#652.

[joostjager]

Is it necessary to make a string out of this? It doesn't need to be mapped to a filename like for fs_store, so maybe it can also simply be a tuple?

[tnull]

I think so, as the HashMap needs to hold an owned value. We could have it be a (String, String, String), but that's worse.

[joostjager]

Is that worse? The string concatenation looks a bit unnecessary. Or make it a struct that is used as the key?

[tnull]

Is that worse? The string concatenation looks a bit unnecessary. Or make it a struct that is used as the key?

It at least requires three individual allocations instead of one? I.e. more clutter on the heap, and probably also some slowdown?

FWIW, I mirrored what we do for the obfuscated key. Unfortunately it's not super easy to just reuse that.

[joostjager]

Not ideal indeed, the heap allocs

[joostjager]

I think we don't need the lock for reading?

[tnull]

Dropped.

[tnull]

Addressed pending comments, probably still need to see why the CI job start hanging again.

[joostjager]

LGTM, can squash

[joostjager]

During final look, I am now again wondering if this function is actually preserving order?

[tnull]

Hmm, good point, it seems it would depend on how tokio exactly schedules the blocking tasks. I considered a few other options, but now simply also followed the tried and true 'write version locking' approach here, as we already use that in VSS and FS stores.

[joostjager]

Hmm but isn't this completely unnecessary because sqlite has its own global lock? With FS and VSS there is the actual possibility of parallel execution.

[tnull]

Hmm but isn't this completely unnecessary because sqlite has its own global lock? With FS and VSS there is the actual possibility of parallel execution.

Well, I first thought so, too, but I think the issue is that tokio gives no guarantee in which order spawned tasks are executed/scheduled. I.e., AFAICT it could happen that we spawn to writes w1, w2 but the task for w2 gets polled first, acquiring the Mutex first. Same goes for the case where multiple writes wait on the same connection lock: say w1 currently holds the Mutex and two more writes w2, w3 would get get queued, AFAIU there is no guarantee that when w1 drops the lock w2 always acquires the lock next.

TLDR: it seems we unfortunately need to do the version dance here, too. Maybe there's an easier mechanism in the SQLite case (e.g., prepare should technically take care of that, but we can't lock the connection Mutex outside of the spawned task as the guard is not Send) we could lean on to guarantee the ordered writes, but applying the same approach seemed simplest for now?

[joostjager]

I just refuse to believe that we need that version dance really for this problem. Isn't block_in_place meant for this? If we ensure that the write has happened before the fn returns, it doesn't matter that during that execution there may be other writes that get processed in a certain order?

[tnull]

Hmm? How do you imagine block_in_place to work here? block_in_place takes a future and drives it to completion, i.e., makes it a blocking operation. For the async KVStore we however exactly need a future, not a blocking operation, which is what spawn_blocking does for us.

[joostjager]

Does it? I saw

pub fn block_in_place<F, R>(f: F) -> R where F: FnOnce() -> R

[tnull]

Does it? I saw

pub fn block_in_place<F, R>(f: F) -> R where F: FnOnce() -> R

Sorry, please replace block_in_place with block_on above. block_in_place is simply a wrapper that spawns a blocking task on the outer runtime context so that the inner block_on call doesn't starve (that is, if it is indeed called on the same runtime, which it isn't always).

[joostjager]

Discussed offline that even though block_in_place may work with caveats, we do prefer to implement sqlite async. And that for that to work, we need the versioning.

[tnull]

Addressed remaining comments and changed base to main. Let me know if I can squash.

[tnull]

CI breakage is unrelated (#654)

[joostjager]

Squash = ok

[tnull]

Squash = ok

Squashed fixups and force-pushed with the following additional changes to address the pending comments:

> git diff-tree -U2 8dada088 c9e3f715
diff --git a/src/io/sqlite_store/mod.rs b/src/io/sqlite_store/mod.rs
index abb45f94..6ba41f71 100644
--- a/src/io/sqlite_store/mod.rs
+++ b/src/io/sqlite_store/mod.rs
@@ -68,9 +68,5 @@ impl SqliteStore {
                &self, primary_namespace: &str, secondary_namespace: &str, key: &str,
        ) -> String {
-               if primary_namespace.is_empty() {
-                       key.to_owned()
-               } else {
-                       format!("{}#{}#{}", primary_namespace, secondary_namespace, key)
-               }
+               format!("{}#{}#{}", primary_namespace, secondary_namespace, key)
        }

@@ -412,30 +408,30 @@ impl SqliteStoreInner {

                self.execute_locked_write(inner_lock_ref, locking_key, version, || {
-               let locked_conn = self.connection.lock().unwrap();
+                       let locked_conn = self.connection.lock().unwrap();

-               let sql = format!("DELETE FROM {} WHERE primary_namespace=:primary_namespace AND secondary_namespace=:secondary_namespace AND key=:key;", self.kv_table_name);
+                       let sql = format!("DELETE FROM {} WHERE primary_namespace=:primary_namespace AND secondary_namespace=:secondary_namespace AND key=:key;", self.kv_table_name);

-               let mut stmt = locked_conn.prepare_cached(&sql).map_err(|e| {
-                       let msg = format!("Failed to prepare statement: {}", e);
-                       io::Error::new(io::ErrorKind::Other, msg)
-               })?;
+                       let mut stmt = locked_conn.prepare_cached(&sql).map_err(|e| {
+                               let msg = format!("Failed to prepare statement: {}", e);
+                               io::Error::new(io::ErrorKind::Other, msg)
+                       })?;

-               stmt.execute(named_params! {
-                       ":primary_namespace": primary_namespace,
-                       ":secondary_namespace": secondary_namespace,
-                       ":key": key,
+                       stmt.execute(named_params! {
+                               ":primary_namespace": primary_namespace,
+                               ":secondary_namespace": secondary_namespace,
+                               ":key": key,
+                       })
+                       .map_err(|e| {
+                               let msg = format!(
+                                       "Failed to delete key {}/{}/{}: {}",
+                                       PrintableString(primary_namespace),
+                                       PrintableString(secondary_namespace),
+                                       PrintableString(key),
+                                       e
+                               );
+                               io::Error::new(io::ErrorKind::Other, msg)
+                       })?;
+                       Ok(())
                })
-               .map_err(|e| {
-                       let msg = format!(
-                               "Failed to delete key {}/{}/{}: {}",
-                               PrintableString(primary_namespace),
-                               PrintableString(secondary_namespace),
-                               PrintableString(key),
-                               e
-                       );
-                       io::Error::new(io::ErrorKind::Other, msg)
-               })?;
-               Ok(())
-        })
        }

[joostjager]

I have to admit that I am a bit nervous about the versioning code, because it is involved in every persistence action. After landing this PR, do we need to do anything extra with regards to testing?

[tnull]

I have to admit that I am a bit nervous about the versioning code, because it is involved in every persistence action. After landing this PR, do we need to do anything extra with regards to testing?

Hmm, well, we have fuzzing test coverage at least on the FilesystemStore variant and by now reviewed essentially the same code/approach 3 times now. But yes, we might want to improve test coverage. Relatedly, we still also need to debug why the VSS CI kept failing (that is, once CI in general is fixed again).