libutil: split retrySQLite into sync and async

also mark the sync version as NeverAsync. a blocking wait on sqlite
locks in a coroutine may never finish if it's a different coroutine
on the same executor that is holding the lock, not another process.

this propagates to the sqlite core interface, but no further. we'll
assume that caches do not block on a database for very long, and we
can't reasonably propagate never-async-ness out of stores unless we
touch everything we'd touch for the async transition, again, twice.
store code already assumes that it can block for however long it'll
feel like that moment. we keep thread pools around for this reason.

Change-Id: I62f77e1ac333cbe2e4e646dbcb1571463f2cf3fc

lix/libexpr/eval-cache.cc

@@ -45,7 +45,7 @@ struct AttrDb
 
         state->db = SQLite(dbPath);
         state->db.isCache();
-        state->db.exec(schema);
+        state->db.exec(schema, always_progresses);
 
         state->insertAttribute = state->db.create(
             "insert or replace into Attributes(parent, name, type, value) values (?, ?, ?, ?)");

lix/libfetchers/cache.cc

@@ -3,6 +3,7 @@
 #include "lix/libutil/async.hh"
 #include "lix/libutil/sync.hh"
 #include "lix/libstore/store-api.hh"
+#include "lix/libutil/types.hh"
 #include "lix/libutil/users.hh"
 
 #include <nlohmann/json.hpp>
@@ -50,7 +51,7 @@ struct CacheImpl : Cache
 
         state->db = SQLite(dbPath);
         state->db.isCache();
-        state->db.exec(schema);
+        state->db.exec(schema, always_progresses);
 
         state->add = state->db.create(
             "insert or replace into Cache(input, info, path, immutable, timestamp) values (?, ?, ?, ?, ?)");

lix/libstore/local-store.cc

@@ -342,20 +342,20 @@ LocalStore::LocalStore(LocalStoreConfig config)
 
         if (curSchema < 8) {
             SQLiteTxn txn = state->db.beginTransaction();
-            state->db.exec("alter table ValidPaths add column ultimate integer");
-            state->db.exec("alter table ValidPaths add column sigs text");
+            state->db.exec("alter table ValidPaths add column ultimate integer", always_progresses);
+            state->db.exec("alter table ValidPaths add column sigs text", always_progresses);
             txn.commit();
         }
 
         if (curSchema < 9) {
             SQLiteTxn txn = state->db.beginTransaction();
-            state->db.exec("drop table FailedPaths");
+            state->db.exec("drop table FailedPaths", always_progresses);
             txn.commit();
         }
 
         if (curSchema < 10) {
             SQLiteTxn txn = state->db.beginTransaction();
-            state->db.exec("alter table ValidPaths add column ca text");
+            state->db.exec("alter table ValidPaths add column ca text", always_progresses);
             txn.commit();
         }
 
@@ -525,7 +525,7 @@ void LocalStore::openDB(DBState & state, bool create)
        all.  This can cause database corruption if the system
        crashes. */
     std::string syncMode = settings.fsyncMetadata ? "normal" : "off";
-    db.exec("pragma synchronous = " + syncMode);
+    db.exec("pragma synchronous = " + syncMode, always_progresses);
 
     /* Set the SQLite journal mode.  WAL mode is fastest, so it's the
        default. */
@@ -538,7 +538,7 @@ void LocalStore::openDB(DBState & state, bool create)
         prevMode = use.getStr(0);
     }
     if (prevMode != mode)
-        db.exec("pragma main.journal_mode = " + mode + ";");
+        db.exec("pragma main.journal_mode = " + mode + ";", always_progresses);
 
     if (mode == "wal" ) {
          /* persist the WAL files when the DB connection is closed.
@@ -547,13 +547,13 @@ void LocalStore::openDB(DBState & state, bool create)
          * journal_size_limit to 2^40 bytes results in the WAL files getting
          * truncated to 0 on exit and limits the on disk size of the WAL files
          * to 2^40 bytes following a checkpoint */
-        db.exec("pragma main.journal_size_limit = 1099511627776;");
+        db.exec("pragma main.journal_size_limit = 1099511627776;", always_progresses);
         db.setPersistWAL(true);
 
         /* Increase the auto-checkpoint interval to 40000 pages.  This
            seems enough to ensure that instantiating the NixOS system
            derivation is done in a single fsync(). */
-        db.exec("pragma wal_autocheckpoint = 40000;");
+        db.exec("pragma wal_autocheckpoint = 40000;", always_progresses);
     }
 
     /* Initialise the database schema, if necessary. */
@@ -561,7 +561,7 @@ void LocalStore::openDB(DBState & state, bool create)
         static const char schema[] =
 #include "schema.sql.gen.hh"
             ;
-        db.exec(schema);
+        db.exec(schema, always_progresses);
     }
 }
 
@@ -812,7 +812,7 @@ void LocalStore::registerDrvOutput(const Realisation & info)
                 (outputId.outputName)
                 .exec();
         }
-    });
+    }, always_progresses);
 }
 
 void LocalStore::cacheDrvOutputMapping(
@@ -885,7 +885,7 @@ std::shared_ptr<const ValidPathInfo> LocalStore::queryPathInfoUncached(const Sto
     return retrySQLite([&]() {
         auto state(_dbState.lockSync(always_progresses));
         return queryPathInfoInternal(*state, path);
-    });
+    }, always_progresses);
 }
 
 
@@ -973,7 +973,7 @@ bool LocalStore::isValidPathUncached(const StorePath & path)
     return retrySQLite([&]() {
         auto state(_dbState.lockSync(always_progresses));
         return isValidPath_(*state, path);
-    });
+    }, always_progresses);
 }
 
 
@@ -994,7 +994,7 @@ StorePathSet LocalStore::queryAllValidPaths()
         StorePathSet res;
         while (use.next()) res.insert(parseStorePath(use.getStr(0)));
         return res;
-    });
+    }, always_progresses);
 }
 
 
@@ -1012,7 +1012,7 @@ void LocalStore::queryReferrers(const StorePath & path, StorePathSet & referrers
     return retrySQLite([&]() {
         auto state(_dbState.lockSync(always_progresses));
         queryReferrers(*state, path, referrers);
-    });
+    }, always_progresses);
 }
 
 
@@ -1028,7 +1028,7 @@ StorePathSet LocalStore::queryValidDerivers(const StorePath & path)
             derivers.insert(parseStorePath(useQueryValidDerivers.getStr(1)));
 
         return derivers;
-    });
+    }, always_progresses);
 }
 
 
@@ -1046,7 +1046,7 @@ LocalStore::queryStaticPartialDerivationOutputMap(const StorePath & path)
                 use.getStr(0), parseStorePath(use.getStr(1)));
 
         return outputs;
-    });
+    }, always_progresses);
 }
 
 std::optional<StorePath> LocalStore::queryPathFromHashPart(const std::string & hashPart)
@@ -1066,7 +1066,7 @@ std::optional<StorePath> LocalStore::queryPathFromHashPart(const std::string & h
         if (s.has_value() && s->starts_with(prefix))
             return parseStorePath(*s);
         return {};
-    });
+    }, always_progresses);
 }
 
 
@@ -1164,7 +1164,7 @@ void LocalStore::registerValidPaths(const ValidPathInfos & infos)
             }});
 
         txn.commit();
-    });
+    }, always_progresses);
 }
 
 
@@ -1524,7 +1524,7 @@ void LocalStore::invalidatePathChecked(const StorePath & path)
         }
 
         txn.commit();
-    });
+    }, always_progresses);
 }
 
 
@@ -1733,7 +1733,7 @@ void LocalStore::addSignatures(const StorePath & storePath, const StringSet & si
         updatePathInfo(*state, *info);
 
         txn.commit();
-    });
+    }, always_progresses);
 }
 
 
@@ -1821,7 +1821,7 @@ std::shared_ptr<const Realisation> LocalStore::queryRealisationUncached(const Dr
     auto maybeRealisation = retrySQLite([&]() {
         auto state(_dbState.lockSync(always_progresses));
         return queryRealisation_(*state, id);
-    });
+    }, always_progresses);
     if (maybeRealisation)
         return std::make_shared<const Realisation>(maybeRealisation.value());
     else

lix/libstore/nar-info-disk-cache.cc

@@ -3,6 +3,7 @@
 #include "lix/libutil/sync.hh"
 #include "lix/libstore/sqlite.hh"
 #include "lix/libstore/globals.hh"
+#include "lix/libutil/types.hh"
 #include "lix/libutil/users.hh"
 #include "lix/libutil/strings.hh"
 
@@ -97,7 +98,7 @@ public:
 
         state->db.isCache();
 
-        state->db.exec(schema);
+        state->db.exec(schema, always_progresses);
 
         state->insertCache = state->db.create(
             "insert into BinaryCaches(url, timestamp, storeDir, wantMassQuery, priority) values (?1, ?2, ?3, ?4, ?5) on conflict (url) do update set timestamp = ?2, storeDir = ?3, wantMassQuery = ?4, priority = ?5 returning id;");
@@ -158,7 +159,7 @@ public:
                     "insert or replace into LastPurge(dummy, value) values ('', ?)")
                     .use()(now).exec();
             }
-        });
+        }, always_progresses);
     }
 
     Cache & getCache(State & state, const std::string & uri)
@@ -219,7 +220,7 @@ public:
 
             txn.commit();
             return ret.id;
-        });
+        }, always_progresses);
     }
 
     std::optional<CacheInfo> upToDateCacheExists(const std::string & uri) override
@@ -234,7 +235,7 @@ public:
                 .wantMassQuery = cache->wantMassQuery,
                 .priority = cache->priority
             };
-        });
+        }, always_progresses);
     }
 
     std::pair<Outcome, std::shared_ptr<NarInfo>> lookupNarInfo(
@@ -278,7 +279,7 @@ public:
             narInfo->ca = ContentAddress::parseOpt(queryNAR.getStr(11));
 
             return {oValid, narInfo};
-        });
+        }, always_progresses);
     }
 
     std::pair<Outcome, std::shared_ptr<Realisation>> lookupRealisation(
@@ -309,7 +310,7 @@ public:
                     "Local disk cache"));
 
             return {oValid, realisation};
-        });
+        }, always_progresses);
     }
 
     void upsertNarInfo(
@@ -349,7 +350,7 @@ public:
                     (hashPart)
                     (time(0)).exec();
             }
-        });
+        }, always_progresses);
     }
 
     void upsertRealisation(
@@ -366,7 +367,7 @@ public:
                 (realisation.id.to_string())
                 (realisation.toJSON().dump())
                 (time(0)).exec();
-        });
+        }, always_progresses);
 
     }
 
@@ -382,7 +383,7 @@ public:
                 (cache.id)
                 (id.to_string())
                 (time(0)).exec();
-        });
+        }, always_progresses);
     }
 };
 

lix/libstore/sqlite.cc

@@ -5,6 +5,7 @@
 #include "lix/libutil/logging.hh"
 #include "lix/libutil/result.hh"
 #include "lix/libutil/signals.hh"
+#include "lix/libutil/types.hh"
 #include "lix/libutil/url.hh"
 
 #include <chrono>
@@ -85,7 +86,7 @@ SQLite::SQLite(const Path & path, SQLiteOpenMode mode)
         sqlite3_trace(db, &traceSQL, nullptr);
     }
 
-    exec("pragma foreign_keys = 1");
+    exec("pragma foreign_keys = 1", always_progresses);
 }
 
 void SQLite::Close::operator()(sqlite3 * db)
@@ -100,11 +101,11 @@ void SQLite::Close::operator()(sqlite3 * db)
 
 void SQLite::isCache()
 {
-    exec("pragma synchronous = off");
-    exec("pragma main.journal_mode = truncate");
+    exec("pragma synchronous = off", always_progresses);
+    exec("pragma main.journal_mode = truncate", always_progresses);
 }
 
-void SQLite::exec(const std::string & stmt)
+void SQLite::exec(const std::string & stmt, NeverAsync)
 {
     retrySQLite([&]() {
         if (sqlite3_exec(db.get(), stmt.c_str(), 0, 0, 0) != SQLITE_OK)

lix/libstore/sqlite.hh

@@ -86,7 +86,7 @@ public:
      */
     void isCache();
 
-    void exec(const std::string & stmt);
+    void exec(const std::string & stmt, NeverAsync = {});
 
     SQLiteStmt create(const std::string & stmt);
 
@@ -223,38 +223,43 @@ kj::Promise<Result<void>> handleSQLiteBusyAsync(const SQLiteBusy & e, std::chron
  * database is busy.
  */
 template<typename F>
-auto retrySQLite(F fun)
+    requires(!requires(F f) { []<typename T>(kj::Promise<T>) {}(f()); })
+auto retrySQLite(F fun, NeverAsync = {})
 {
     auto nextWarning = std::chrono::steady_clock::now() + std::chrono::seconds(1);
 
-    if constexpr (requires (F f) { []<typename T>(kj::Promise<Result<T>>){}(f()); }) {
-        return [](std::chrono::time_point<std::chrono::steady_clock> nextWarning, F fun) -> decltype(fun()) {
-            while (true) {
-                kj::Promise<Result<void>> handleBusy{nullptr};
-                try {
-                    if constexpr (std::is_same_v<decltype(fun()), kj::Promise<Result<void>>>) {
-                        LIX_TRY_AWAIT(fun());
-                        co_return result::success();
-                    } else {
-                        co_return LIX_TRY_AWAIT(fun());
-                    }
-                } catch (SQLiteBusy & e) {
-                    handleBusy = handleSQLiteBusyAsync(e, nextWarning);
-                } catch (...) {
-                    co_return result::current_exception();
-                }
-                LIX_TRY_AWAIT(handleBusy);
-            }
-        }(nextWarning, std::move(fun));
-    } else {
-        while (true) {
-            try {
-                return fun();
-            } catch (SQLiteBusy & e) {
-                handleSQLiteBusy(e, nextWarning);
-            }
+    while (true) {
+        try {
+            return fun();
+        } catch (SQLiteBusy & e) {
+            handleSQLiteBusy(e, nextWarning);
         }
     }
 }
 
+template<typename F>
+    requires requires(F f) { []<typename T>(kj::Promise<Result<T>>) {}(f()); }
+auto retrySQLite(F fun)
+{
+    return [](F fun) -> decltype(fun()) {
+        auto nextWarning = std::chrono::steady_clock::now() + std::chrono::seconds(1);
+
+        while (true) {
+            kj::Promise<Result<void>> handleBusy{nullptr};
+            try {
+                if constexpr (std::is_same_v<decltype(fun()), kj::Promise<Result<void>>>) {
+                    LIX_TRY_AWAIT(fun());
+                    co_return result::success();
+                } else {
+                    co_return LIX_TRY_AWAIT(fun());
+                }
+            } catch (SQLiteBusy & e) {
+                handleBusy = handleSQLiteBusyAsync(e, nextWarning);
+            } catch (...) {
+                co_return result::current_exception();
+            }
+            LIX_TRY_AWAIT(handleBusy);
+        }
+    }(std::move(fun));
+}
 }

tests/unit/libstore/nar-info-disk-cache.cc

@@ -4,6 +4,7 @@
 #include <rapidcheck/gtest.h>
 #include "lix/libstore/sqlite.hh"
 #include "lix/libstore/temporary-dir.hh"
+#include "lix/libutil/types.hh"
 #include <sqlite3.h>
 
 
@@ -59,7 +60,7 @@ TEST(NarInfoDiskCacheImpl, create_and_read) {
         }
 
         // Pretend that the caches are older, but keep one up to date, as "background noise"
-        db.exec("update BinaryCaches set timestamp = timestamp - 1 - 7 * 24 * 3600 where url <> 'https://xyz';");
+        db.exec("update BinaryCaches set timestamp = timestamp - 1 - 7 * 24 * 3600 where url <> 'https://xyz';", always_progresses);
 
         // This shows that the in-memory cache works
         {