diff --git a/Cargo.lock b/Cargo.lock index 436299524e..1b951752e8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -224,22 +224,13 @@ dependencies = [ "zstd-safe 7.2.0", ] -[[package]] -name = "async-lock" -version = "2.8.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "287272293e9d8c41773cec55e365490fe034813a2f172f502d6ddcf75b2f582b" -dependencies = [ - "event-listener 2.5.3", -] - [[package]] name = "async-lock" version = "3.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff6e472cdea888a4bd64f342f09b3f50e1886d32afe8df3d663c01140b811b18" dependencies = [ - "event-listener 5.3.1", + "event-listener", "event-listener-strategy", "pin-project-lite", ] @@ -1920,12 +1911,6 @@ dependencies = [ "windows-sys 0.52.0", ] -[[package]] -name = "event-listener" -version = "2.5.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0" - [[package]] name = "event-listener" version = "5.3.1" @@ -1943,7 +1928,7 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0f214dc438f977e6d4e3500aaa277f5ad94ca83fbbd9b1a15713ce2344ccc5a1" dependencies = [ - "event-listener 5.3.1", + "event-listener", "pin-project-lite", ] @@ -3010,7 +2995,7 @@ dependencies = [ "aes", "anyhow", "arbitrary", - "async-lock 2.8.0", + "async-lock", "async-recursion", "async-stream", "async-tempfile", @@ -3423,12 +3408,12 @@ version = "0.12.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32cf62eb4dd975d2dde76432fb1075c49e3ee2331cf36f1f8fd4b66550d32b6f" dependencies = [ - "async-lock 3.4.0", + "async-lock", "async-trait", "crossbeam-channel", "crossbeam-epoch", "crossbeam-utils", - "event-listener 5.3.1", + "event-listener", "futures-util", "once_cell", "parking_lot", diff --git a/docs/ADMIN_API.md b/docs/ADMIN_API.md index 651f43ebae..c15404aa13 100644 --- a/docs/ADMIN_API.md +++ b/docs/ADMIN_API.md @@ -53,6 +53,25 @@ Both importers produce the same data. Known differences: - invalid UTF-8 or NUL bytes yield `400` (buffered: `500`), and statements that return rows are executed with their rows discarded (buffered: `500`). +Creation is all-or-nothing. The namespace becomes visible, and its configuration is persisted, +only after it has been fully set up (including the dump import). If the request fails, is +cancelled by the client, or the server dies while it is in flight, the namespace either does not +exist (its name can be created again right away — a failed attempt is fully undone before the +error is returned) or is complete — never partially imported. Requests addressed to a namespace +while it is being created wait for the outcome. Directories of creations the server died in the +middle of are discarded at the next start (`/.discarded/` holds them until they are +deleted). For callers: + +- `2xx`: the namespace is complete; +- any other outcome, including a lost connection: retry the same request; a + `400 Namespace already exists` on the retry means the earlier attempt did complete. + +In single-namespace mode (`--disable-namespaces`) creating the default namespace is a +configuration upsert and always succeeds, except with a `dump_url`: a dump can only be imported +into a namespace that does not exist yet, so that request is rejected like any other. + +See `docs/ATOMIC_NAMESPACE_CREATE_DESIGN.md`. + ```HTTP DELETE /v1/namespaces/:namespace ``` diff --git a/docs/ATOMIC_NAMESPACE_CREATE_DESIGN.md b/docs/ATOMIC_NAMESPACE_CREATE_DESIGN.md new file mode 100644 index 0000000000..fc3489e44d --- /dev/null +++ b/docs/ATOMIC_NAMESPACE_CREATE_DESIGN.md @@ -0,0 +1,349 @@ +# Atomic namespace creation for `libsql-server` + +Stacked on `tszymczyszyn/streaming-dump-importer` (PR #51). Fixes the pre-existing lifecycle gap described in PR #51's design doc §13 and the review walkthrough: a failed, cancelled or crashed `POST /v1/namespaces/:ns/create` could leave a *ghost* namespace (metastore row without data), a directory without a row, or both. Sections marked *as built* note where the implementation differs from the sketches; §9 records what the adversarial review of PR #53 changed. + +--- + +## 1. Problem + +![Before and after](images/atomic-namespace-create-before-after.png) + + +`NamespaceStore::create` persists the namespace config **first** and only then runs the configurator's `setup` (directory, logger, connections, dump import). Three resources are mutated without a common commit point: + +| Resource | Written at | Cleaned on error | Cleaned on cancellation | Cleaned after crash | +|---|---|---|---|---| +| metastore row (`namespace_configs`) | start of `create` | **no** | **no** | n/a (persisted) | +| `dbs//` + `.sentinel` + SQLite/WAL/log files | during `setup` | primary only, if the dir was fresh | **no** (future dropped, `Err` branch never runs) | **no** | +| cache entry (`NamespaceEntry`) | end of `load_namespace` | n/a | n/a | n/a | + +Consequences: a failed create reserves the name forever (`NamespaceAlreadyExist` on retry) while user traffic lazily materialises an **empty** database; a cancelled create can leave any prefix of the resources; a client cannot tell from a non-200 whether the namespace now exists. `SchemaConfigurator::setup` has no cleanup at all. `fork` has a drop guard for the row but uses `block_in_place`, which panics on a current-thread runtime. + +Both dump importers run inside this sequence; the problem is independent of PR #51. + +## 2. Goals + +- **Contract:** after `create` returns a non-2xx, or the connection is lost, the namespace is either *absent* (no row, no data, name reusable) or *complete* (row persisted, data complete, serving). Never partial. +- Cancellation-safe (dropping the request future cleans up) and crash-tolerant (a process crash mid-import never blocks a retry). +- No lock-step changes to clients, replication, the metastore schema or the `DatabaseConfig` protobuf. +- Small: reuse the pattern `fork` already uses; one mechanism for `create` and `fork`. + +Non-goals: a server-side import timeout; idempotency keys / async create jobs; bottomless (not used in our deployment; untested with this change). The moka eviction hazard shared with `reset`/`fork` turned out to need fixing (§8). + +## 3. Design in one paragraph + +![Components and ownership](images/atomic-namespace-create-components.png) + +*Who owns what: numbered arrows are the happy path, dashed red the undo. Live board (Shopify-internal, editable): https://drawr.quick.shopify.io/b/GNUB8o_56hhfFGxv — page "Before / after" contrasts the old and new sequence.* + + +Make the **metastore write the commit point** and make everything before it reversible. `create` reserves the name by write-locking its cache entry, puts the config in the metastore's *in-memory* map only (so `setup` can read it), runs `setup`, and only on success **flushes the row** and installs the namespace. A `Reservation` drop guard undoes the in-memory config if the future ends any other way — error *or* cancellation — and releases the lock only after that, so waiters never observe a half-built namespace. The filesystem gets the same treatment one layer down: `setup` wraps a brand-new directory in a `FreshDir` guard that removes it on drop. For crashes, a brand-new directory carries an `.incomplete` marker until the row is flushed; the next `create` of that name discards a marked directory. + +``` +create(name, restore, config) +│ +├─ Reservation::acquire(name) lock cache entry; reject if loaded or row exists +├─ handle = metadata.handle(name) ─┐ in-memory only; nothing on disk +├─ handle.store_and_maybe_flush(cfg, false) ┘ +├─ configurator.discard_incomplete(name) remove `dbs/` if it carries `.incomplete` +├─ ns = make_namespace(...) setup: FreshDir::begin → mkdir + marker … import … keep() +├─ handle.flush() ◄── COMMIT POINT: row persisted +├─ ns.mark_complete() remove `.incomplete` (best effort) +└─ reservation.publish(ns) entry := Some(ns); lock released + any Err / drop above ⇒ Reservation::drop ⇒ metadata.remove(name), then release lock + any Err / drop inside setup ⇒ FreshDir::drop ⇒ remove_dir_all(dbs/) +``` + +## 4. Components + +### 4.1 `Reservation` (store.rs) — replaces `fork`'s ad-hoc `Bomb` + +```rust +/// A namespace name reserved for a creation or fork in progress. +/// +/// Holding it holds the name's cache entry write-locked: user requests, a second create, +/// destroy and shutdown all wait until the outcome is known. `publish` installs the finished +/// namespace. Dropping it unpublished (error or cancelled request) removes the in-memory +/// config and only then releases the lock, so waiters see "doesn't exist", never a partial one. +struct Reservation { + guard: Option>>, + metadata: MetaStore, + name: NamespaceName, +} + +impl Reservation { + async fn acquire(store: &NamespaceStore, name: NamespaceName) -> crate::Result { + let entry = store.inner.store.get_with(name.clone(), async { Default::default() }).await; + let guard = entry.write_arc().await; + if guard.is_some() || store.inner.metadata.exists(&name).await { + return Err(Error::NamespaceAlreadyExist(name.to_string())); + } + Ok(Self { guard: Some(guard), metadata: store.inner.metadata.clone(), name }) + } + + fn publish(mut self, ns: Namespace) { + self.guard.take().expect("published once").replace(ns); + } +} + +impl Drop for Reservation { + fn drop(&mut self) { + let Some(guard) = self.guard.take() else { return }; + let (metadata, name) = (self.metadata.clone(), self.name.clone()); + tracing::warn!(namespace = %name, "namespace creation did not complete; discarding it"); + let cleanup = move || { + if let Err(e) = metadata.remove(name.clone()) { + tracing::error!(namespace = %name, "failed to discard namespace config: {e}"); + } + drop(guard); // release only once the name no longer exists + }; + match tokio::runtime::Handle::try_current() { + Ok(rt) => { rt.spawn_blocking(cleanup); } + Err(_) => cleanup(), + } + } +} +``` + +Why `write_arc`: the guard must outlive the `create` future (it moves into the cleanup task). Requires `async-lock = "3"` in `libsql-server/Cargo.toml` (3.4.0 is already in the lockfile transitively; `proxy.rs`'s `RwLockUpgradableReadGuard::upgrade` is unchanged in 3.x). + +`MetaStore::remove` takes blocking locks, hence `spawn_blocking` (no `block_in_place`, so it also works on turmoil's current-thread runtime). It deletes the row if one exists and the in-memory entry; both are what we want in every abort path, including "flush succeeded but publish didn't" (then the row is deleted again and the complete directory becomes a marked orphan, see §6). + +### 4.2 `create` (store.rs) + +```rust +pub async fn create(&self, namespace, restore_option, db_config) -> crate::Result<()> { + shutdown check — unchanged + shared_schema_name ⇒ fork(…) — unchanged (fork now uses Reservation, §4.6) + + // Single-namespace mode / default namespace: `create` is an idempotent config upsert of a + // namespace that always exists. Keep today's path for it, but only for `Latest`: creating + // *from a dump* must never silently skip the import because the namespace is already loaded + // (today it does), so dump creates always take the reserving path below. + if (self.inner.allow_lazy_creation || namespace == NamespaceName::default()) + && matches!(restore_option, RestoreOption::Latest) + { + let handle = self.inner.metadata.handle(namespace.clone()).await; + handle.store(Arc::new(db_config)).await?; + self.load_namespace(&namespace, handle, restore_option).await?; + return Ok(()); + } + + let reservation = Reservation::acquire(self, namespace.clone()).await?; + let handle = self.inner.metadata.handle(namespace.clone()).await; + handle.store_and_maybe_flush(Some(Arc::new(db_config)), false).await?; + self.get_configurator(&handle.get()).discard_incomplete(&namespace).await?; + let ns = self.make_namespace(&namespace, handle.clone(), restore_option).await?; + handle.flush().await?; // commit point + if let Err(e) = ns.mark_complete().await { // cosmetic from here on + tracing::warn!(namespace = %namespace, "could not remove creation marker: {e}"); + } + reservation.publish(ns); + Ok(()) +} +``` + +Ordering that matters: `acquire` before `handle()` (`handle()` inserts into the in-memory map, which is what `exists()` reads — so the name becomes visible only while it is already locked); `flush` before `publish` (durable before visible); `publish` consumes the reservation (no double cleanup). + +### 4.3 `FreshDir` (configurator/helpers.rs) — cancellation-safe directory + +> *As built (review):* `begin` is synchronous (guard constructed before anything is created, no await between `mkdir` and the marker), and `Drop` does not delete in place: it calls `incomplete::discard_dir`, which atomically **renames the directory into `/.discarded/-`** and deletes the tombstone in the background. The path under `dbs/` is therefore free the instant the guard drops, a retry can never race a half-finished `remove_dir_all`, and files a still-running import executor recreates land in the tombstone. The sketch below shows the original idea. + +```rust +/// Directory of a namespace that does not exist yet. Removed again if dropped before `keep`: +/// on error, or when the request creating the namespace is cancelled. +pub(super) struct FreshDir { path: Option> } + +impl FreshDir { + /// `None` if a namespace already lives at `db_path`. + pub(super) async fn begin(db_path: &Arc) -> crate::Result> { + if db_path.try_exists()? { return Ok(None); } + tokio::fs::create_dir_all(db_path).await?; + tokio::fs::File::create(db_path.join(INCOMPLETE_MARKER)).await?; + Ok(Some(Self { path: Some(db_path.clone()) })) + } + pub(super) fn keep(mut self) { self.path.take(); } +} + +impl Drop for FreshDir { + fn drop(&mut self) { + let Some(path) = self.path.take() else { return }; + let rm = async move { + match tokio::fs::remove_dir_all(&*path).await { + Ok(()) | Err(e) if e.kind() == ErrorKind::NotFound => {} + Err(e) => tracing::error!(path = %path.display(), "failed to remove unfinished namespace directory: {e}"), + } + }; + match tokio::runtime::Handle::try_current() { + Ok(rt) => { rt.spawn(rm); } + Err(_) => { let _ = std::fs::remove_dir_all(&*path); } + } + } +} +``` + +Used by `PrimaryConfigurator::setup` and `SchemaConfigurator::setup` (which gains cleanup it never had): + +```rust +let db_path: Arc = self.base.base_path.join("dbs").join(name.as_str()).into(); +let fresh = FreshDir::begin(&db_path).await?; +let ns = self.try_new_primary(…).await?; // on Err or drop, `fresh` removes the directory +if let Some(fresh) = fresh { fresh.keep(); } +Ok(ns) +``` + +`try_new_primary`'s own `create_dir_all` becomes redundant for the fresh case and stays harmless for existing directories. The `remove_dir_all`-on-error branch in `setup` is replaced by the guard (drop order: the `try_new_primary` future and its `JoinSet` are dropped before `fresh`, so background tasks are aborted before the directory goes). + +### 4.4 `.incomplete` marker — crash tolerance + +> *As built (review):* the protocol lives in `namespace/incomplete.rs` (`INCOMPLETE_MARKER`, `create_marked_dir`, `discard_dir`, `discard_if_incomplete`, `sweep`). `sweep` runs at startup (`lib.rs`, after the store is built): it deletes leftover tombstones and discards marked directories of names the metastore does not know. `fork` writes the marker into its temporary directory before renaming it into place, so an unfinished fork is discarded like any other unfinished creation. The order at the end of a creation is **flush → publish → remove marker** (not flush → mark → publish): a marker that outlives publication is harmless, a complete directory without one is not. + +- `pub(crate) const INCOMPLETE_MARKER: &str = ".incomplete";` in `namespace/mod.rs`. +- Created by `FreshDir::begin` for every brand-new directory. +- Removed by `Namespace::mark_complete(&self)` (`remove_file(self.path.join(MARKER))`, `NotFound` ok), called (a) by `create` after `flush`, (b) by `load_namespace`'s init after `make_namespace` — a namespace opened from the metastore is complete by definition, so lazily loaded namespaces (and single-namespace-mode auto-creation) never keep a marker. +- Acted upon only for a name the metastore does **not** know: by `ConfigureNamespace::discard_incomplete(name)` (new trait method, default no-op; primary and schema delegate to `incomplete::discard_if_incomplete`), which `create` calls after `acquire` — i.e. with no row and nothing loaded — and by the startup `sweep` under the same condition. That is the only situation in which a marked directory is unambiguously garbage. Lazy loads never act on it, so a stale marker on a published namespace (crash between `flush` and unmark) is removed, never acted upon. The metastore's filesystem recovery (`maybe_recover_from_fs`, empty metastore only) skips marked directories rather than adopting them. +- `MetaStoreInner::maybe_recover_from_fs` (metastore empty ⇒ adopt `dbs/*`) skips directories carrying the marker, so a crash during the very first creation on a fresh server does not get adopted as a namespace at the next start. + +A directory **without** a marker and without a row keeps today's semantics (`Dump` ⇒ `LoadDumpExistingDb`; `Latest` ⇒ adopted). Such directories can only come from pre-marker servers or manual intervention; being destructive there is not worth it. + +### 4.5 `Namespace::mark_complete` (namespace/mod.rs) + +```rust +pub(crate) async fn mark_complete(&self) -> std::io::Result<()> { + match tokio::fs::remove_file(self.path.join(INCOMPLETE_MARKER)).await { + Err(e) if e.kind() != ErrorKind::NotFound => Err(e), + _ => Ok(()), + } +} +``` + +### 4.6 `fork` (store.rs) + +Replace the `Bomb` + manual `to_lock` with `Reservation::acquire(to)`; order becomes `configurator.fork(…)` → `handle.flush()` → `reservation.publish(ns)` (today it publishes *then* flushes, so a flush failure leaves a running namespace without a row). Net code removal; same semantics otherwise. `ForkTask`'s temp-dir + rename stays, plus the marker (§4.4). *As built:* the source namespace is looked up with the side-effect-free `MetaStore::existing_handle` (§9). + +## 5. Flow for a dump import (streaming or buffered) + +1. Admin handler resolves `dump_url` (errors here touch nothing). +2. `create` → `Reservation::acquire` (`NamespaceAlreadyExist`, HTTP 400, if the name is loaded or has a row). +3. In-memory config; `discard_incomplete` (removes a crashed previous attempt, if any). +4. `setup`: `FreshDir::begin` (mkdir + marker) → logger, WAL, connection maker → `load_dump` → `keep()`. +5. `flush` — the namespace now exists durably. +6. `mark_complete`, `publish` — the namespace is now visible; waiters proceed. + +## 6. Failure matrix + +| Interrupted at | Error (returned) | Cancellation (future dropped) | Crash (process dies) | +|---|---|---|---| +| 1 | nothing to undo | nothing to undo | nothing | +| 2–3 (before `setup`) | `Reservation` removes in-memory config | same | nothing persisted; nothing on disk | +| 4, inside `setup` (incl. import) | `FreshDir` removes dir; `Reservation` removes config; executor rolls back SQLite | same (importer executor is uncancelled: closes channel → ROLLBACK → connection released; dir removal races with it benignly on POSIX, see §9) | marked orphan dir, no row ⇒ invisible; next `create` discards it | +| between `setup` and `flush` | config removed; **complete dir, marked** ⇒ next `create` discards and redoes | same | same | +| `flush` fails | row never written (single SQLite statement); as above | — | — | +| between `flush` and `publish` | `Reservation` deletes the row again; marked complete dir ⇒ discarded with it | same | row + marker + complete data ⇒ namespace **exists and serves**; marker removed on first lazy load | +| between `publish` and marker removal | — (published; a stale marker is harmless) | same | row + marker + complete data ⇒ serves; marker removed on first lazy load | +| after `publish` | — | response may be lost ⇒ client retries ⇒ `NamespaceAlreadyExist` ⇒ by the contract, complete | same | + +Client rule: **2xx ⇒ complete; anything else ⇒ retry; `NamespaceAlreadyExist` on retry ⇒ complete.** No inspection of tables needed. (For namespaces *created* through this path. The single-namespace-mode config upsert of the default namespace, and ghost rows left by older binaries, are outside it.) + +## 7. Concurrency + +- **Second `create` of the same name:** waits on the write lock; then `AlreadyExist` (first succeeded) or proceeds (first failed). Previously: `AlreadyExist` even after the first failed. +- **User request during creation (`with`)**: `exists()` is true (in-memory config), `try_get_with` finds the locked entry, `read()` waits; then sees the namespace or `None ⇒ NamespaceDoesntExist`. Same waiting behaviour as during `fork`/`reset` today. +- **`destroy` during creation:** `metadata.remove` succeeds (in-memory only), then waits on the pinned entry's lock; when creation finishes, `install` notices the config is gone and hands the namespace to that entry, `destroy` takes and destroys it. Net: created then destroyed, in that order (`destroy_during_creation_destroys_the_result`). +- **Store shutdown during creation:** `shutdown` waits on the entry lock, then shuts the finished namespace down normally. +- **`handle()` side effect:** `MetaStore::handle` inserts a default config into the in-memory map (pre-existing). Keeping `acquire` strictly before `handle()` is what makes the reservation airtight; documented in code. + +## 8. Cache eviction during an operation (*as built*: fixed, not merely inherited) + +The reservation is the cache entry's lock, but moka knows nothing about that lock. If the cache evicts the entry during a long operation (capacity pressure: `--max-active-namespaces`, default 100, is reached and more namespaces get loaded; the 24 h idle expiry is irrelevant), three things used to go wrong — for `fork` and `reset` since before this change, and for `create` once it adopted the same reservation (previously `create` ran inside moka's `try_get_with`, which deduplicates in-flight inits and is immune): + +1. **Split-brain.** The cache slot is empty, so a concurrent lookup (`exists()` is true: the config is in memory) runs a lazy load and opens the directory the operation is writing to — a second live instance on the same files. +2. **Stalled maintenance.** moka 0.12 has no background thread: maintenance runs inline inside whatever cache operation trips it, and it *awaits* async eviction listeners. The listener did `entry.write().await`, i.e. it waited for the operation to finish. The unrelated request that happened to run maintenance hung for the duration; once the write-op channel filled, every namespace load on the node blocked. +3. **Self-deadlock.** If the operation holding the lock touched the cache itself (`fork` does: `load_namespace(from)`), maintenance could pick up the eviction of its own entry and wait on its own lock. The first version of the unit test below reproduced exactly this. + +Fix, in three parts: + +- **Pin.** `NamespaceStoreInner::in_flight: Mutex>`. `Reservation::acquire` (create, fork) and `reset` pin the locked entry; `load_namespace` consults the map *before* `try_get_with` and waits on the pinned entry instead of loading. If the entry ends up empty — the operation failed, or published elsewhere — the lookup falls through: `NamespaceDoesntExist` if the metastore no longer knows the name, otherwise the regular cache path. Unpin happens after the config removal (abort) or after the namespace is visible (publish), so no lookup can slip into a lazy load in between. `destroy` consults the map too. +- **Non-blocking eviction listener.** The listener spawns the shutdown on its own task and returns immediately, so maintenance never waits for an operation in progress. +- **Publish into a fresh entry when evicted.** `NamespaceStore::install` checks (by `Arc::ptr_eq`) that the locked entry is still the cache's entry for the name; if not, the namespace goes into a fresh entry and the orphaned one stays empty, so the spawned listener finds nothing to shut down and waiters fall through to the cache. Used by `create`, `fork` and `reset`. + +Unit tests (`namespace::store::test`) build a real `NamespaceStore`, start a streaming import fed from a channel, `invalidate` the reserved entry (same code path as eviction, including the listener) and check that a concurrent lookup waits and then sees the imported data (or `NamespaceDoesntExist` if the import fails), that the namespace ends up in the cache, and that nothing stays pinned. Without the pin the lookup returns early with an empty database; without the non-blocking listener the test deadlocks. + +## 9. Adversarial review of PR #53 — what changed (*as built*) + +Reviewers (correctness, architecture, testing, operations) plus a manual pass found the following; all fixed in the fourth commit. + +| Sev | Finding | Resolution | +|---|---|---| +| HIGH | Directory removal was spawned and unordered against releasing the name: a retry "right away" (the documented contract) could race `remove_dir_all`; tests hid it with `wait_until_gone`. | `discard_dir`: atomic rename into `.discarded/` (sync, in `Drop`), background delete, startup `sweep`. Error paths now `await` the reservation's cleanup (`Reservation::abort`); `Drop` is for cancellation only. `retry_right_after_failure_starts_clean` asserts the path and the name are free the moment the error is returned. | +| HIGH | `MetaStore::remove` returns early (`HasLinkedDbs`, `PendingMigrationOnSchema`) **before** dropping the in-memory config; reachable for a linked-DB create whose flush fails ⇒ ghost. | `MetaStore::discard_blocking`: unconditional, goes through the metastore's own change queue (so it is also ordered after a config store that was still queued when the request was cancelled). `discard_ignores_the_guards_of_remove`. | +| HIGH | `install`'s `Arc::ptr_eq` fallback could not tell eviction from `destroy`: destroy-during-create re-inserted a live namespace whose directory destroy had just removed; name wedged. | Fallback re-checks `exists()`: config gone ⇒ hand the namespace to the old entry (destroy is waiting on it); `destroy` prefers the pinned entry. `destroy_during_creation_destroys_the_result`. | +| HIGH | Cancellation between `flush` and `publish` after `mark_complete` ⇒ row deleted, complete dir **unmarked** ⇒ `LoadDumpExistingDb` forever. | Order is flush → publish → remove marker. | +| HIGH | Fork's destination (renamed temp dir) carried no marker ⇒ a cancelled fork left an unmarked orphan and `ENOTEMPTY` on retry. | Marker written into the temp dir. | +| HIGH (found while testing) | A rejected `acquire` (`AlreadyExist` for an existing but unloaded namespace, e.g. a retry after restart) left the empty entry its `get_with` inserted ⇒ every lookup found the empty entry instead of loading ⇒ namespace unreachable until eviction. `checkpoint` has had the same latent bug. | `acquire` invalidates the empty entry it created; `load_namespace` treats an empty, unlocked entry as a miss and reloads (self-heals `checkpoint`/failed-`reset` leftovers too). `marker_on_published_namespace_is_ignored` creates before any lookup after a restart. | +| MED | ABBA: `remove` takes `conn` then `configs`; `process`'s flush branch held `configs` across `try_process` (`conn`). Now on every create. | `configs` released before `try_process`. | +| MED | Spawning the eviction listener unconditionally dropped moka's per-key serialisation for *normal* evictions (reload could overlap the old instance's shutdown). | Listener awaits inline unless the entry is pinned (`in_flight` shared with the closure). | +| MED | `shutdown` waited on an in-flight create's lock; `create` never re-checked `has_shutdown`. | Pinned entries skipped at shutdown (warn); `has_shutdown` re-checked before the commit point. | +| MED | `with()` called `MetaStore::handle`, which *inserts* into the map `exists()` reads: a lookup racing the end of a failed creation resurrected the name. | `MetaStore::existing_handle` (non-inserting) for namespaces that must exist; lazy init re-checks `exists()`. | +| MED | `FreshDir::begin` had two awaits before the guard existed. | Synchronous; guard first. | +| MED | Cancel tests could fire before the creation started (turmoil latency) and so pass vacuously; the restart test never reached the `\|\| exists` guard. | Cancel once `dbs/foo/.incomplete` exists; create before any lookup after restart; "usable afterwards" assertion restored. | +| LOW | Marked orphans were only reclaimed by a `create` of the same name; `maybe_recover_from_fs` skip untested; success not logged. | Startup `sweep`; `recovery_from_fs_skips_marked_directories`, `sweep_cleans_leftovers_of_a_crash`; `info!("namespace created")` with elapsed, `warn!` with the error on failure, distinct message on cancellation. | + +LOW items closed in the follow-up commit: `publish` keeps the guard inside the reservation across `install`'s awaits, so a cancellation there undoes the creation instead of leaking the pin; `remove_incomplete_marker` stats before taking the blocking-pool hop (every cold load passes through it); a failure to remove the marker after publish is logged at `error`; `FreshDir`'s log is `info` (every failed create, including client errors, passes through it); `acquire` logs when it has to wait; store-level `aborted_creation_is_undone` test (`JoinHandle::abort` mid-import); the restart test drops the first server before starting the second; the unmarked-directory test asserts the outcome. + +Also closed: the marker and its directory entry are fsynced on creation (`mark_incomplete`), so the marker is durable before anything else in the directory can be; `checkpoint` no longer mints an empty cache entry for an unloaded namespace (the pre-existing source of the same wedge); `ForkTask::fork` only removes a failed fork's destination if it carries the marker. + +Not changed: after an *abort* the (empty) cache entry stays until a lookup heals it or the cache evicts it — it counts toward capacity but is never served; a metastore rebuilt from the filesystem (`destroy_on_error` + `allow_recover_from_fs`) in the milliseconds between a namespace's flush and its unmark would skip that namespace. Bottomless is not used in our deployment and this change is untested with it. + +**Upgrade / downgrade.** Ghost rows and unmarked orphan directories left by failed creates on *older* binaries are untouched (everything here is marker-gated); for them the old remedy stays: `DELETE /v1/namespaces/:ns`, then create again. Downgrading the binary is safe: the metastore schema and the config protobuf are unchanged; an old binary ignores `.incomplete` and `.discarded/` (it may adopt a marked orphan via `maybe_recover_from_fs` on an empty metastore, as it always could with any leftover directory). + +## 10. Interaction with the streaming importer + +On cancellation the reader future is dropped, the executor thread sees the closed channel, rolls back and drops its connection — concurrently with the directory being discarded. The directory is renamed, not deleted in place, so whatever the executor still writes lands in the tombstone; on POSIX, renaming a directory with open files in it is harmless. The tombstone is deleted in the background, or by `sweep` at the next start. + +## 11. Observability + +- `warn!` on every abort path (`Reservation`: "namespace creation did not complete; discarding it"; `FreshDir`: "… removing its directory"; `discard_incomplete`: "discarding namespace directory left by an interrupted creation"), with the namespace / path. +- Counters `libsql_server_namespace_create_aborted` and `libsql_server_namespace_create_discarded_incomplete` (*as built:* no `stage` label — the logs carry the detail). +- The contract of §6 is stated in `docs/ADMIN_API.md` under namespace creation. + +## 12. Tests (turmoil, `tests/namespaces/lifecycle.rs` and `dumps.rs`) + +*As built.* 9 of the 11 integration tests fail on the unfixed code; the other two pin behaviour that must hold both before and after. + +1. `failed_create_leaves_no_trace{,_streaming,_shared_schema}`: invalid dump ⇒ 400; `GET /v1/namespaces/foo/config` ⇒ 404 (was 200); hrana ⇒ "doesn't exist" (was "no such table"); `dbs/foo` gone; `create` again with a valid dump ⇒ 200, data present, no marker. +2. `cancelled_create_leaves_no_trace{,_streaming}` (replaces `streaming_cancelled_request_rolls_back`, which tolerated "OK or BAD_REQUEST" on retry): request abandoned mid-transfer ⇒ config 404, hrana "doesn't exist", directory gone, retry from the same dump ⇒ 200 with all rows. +3. `concurrent_requests_wait_for_creation`: a second `create` and a user query issued during a ~2 s import wait; afterwards the second gets `already exists` and the query sees all rows. `concurrent_create_succeeds_after_failure`: the import fails ⇒ the waiting query gets "doesn't exist" and a new `create` succeeds. +4. `crash_remnant_is_discarded`: `dbs/foo/{.incomplete,.sentinel,data,wallog}` exists before the server starts ⇒ not adopted by the metastore (config 404), `create foo` from a dump ⇒ 200. `unmarked_directory_is_left_alone`: a marker-less `dbs/foo/keep.txt` survives a `create`. `marker_on_published_namespace_is_ignored`: marker added to a published namespace ⇒ `create` ⇒ `already exists`, data served; after a restart the marker is gone. +5. `single_namespace_mode_rejects_dump_into_existing_namespace`: config upsert of `default` still works; `create default` from a dump ⇒ `already exists` (was: 200, dump silently skipped). +6. Unit (`configurator::helpers::test`): `FreshDir` removed on drop and when its enclosing future is dropped, kept on `keep()`; `discard_incomplete` removes only marked directories. +7. `fork_namespace`, `shared_schema::*` and all dump tests unchanged. +8. `namespace::store::test::lookup_{waits_for_creation,sees_failure_of_creation}_even_if_the_entry_was_evicted` (§8), `destroy_during_creation_destroys_the_result`, `retry_right_after_failure_starts_clean`, `discard_ignores_the_guards_of_remove`; `namespace::incomplete::test::*` (§9). + +Slow-transfer tests use `make_slow_dump_store` (64-byte chunks, 100 ms apart): with turmoil's random 0–100 ms message latency, concurrent requests are issued 500 ms in and the import lasts ~2 s. + +## 13. Scope + +| File | Change | +|---|---| +| `libsql-server/Cargo.toml` | `async-lock = "3"` | +| `namespace/store.rs` | `Reservation` (+ `abort`); `create`/`fork` reserving paths; `load_namespace` (pinned entries, empty-entry healing, `mark_complete`); `in_flight` pin map, pin-aware eviction listener, `install`, `shutdown` (§8, §9) | +| `namespace/incomplete.rs` | marker, `discard_dir` (tombstones), `discard_if_incomplete`, `sweep` | +| `namespace/meta_store.rs` | `ChangeMsg` enum with `Discard`, `discard_blocking`, `existing_handle`, lock-order fix | +| `lib.rs` | startup `sweep` | +| `namespace/mod.rs` | `remove_incomplete_marker`, `Namespace::mark_complete` | +| `namespace/configurator/mod.rs` | `ConfigureNamespace::discard_incomplete` (default no-op) | +| `namespace/configurator/helpers.rs` | `FreshDir`, `discard_incomplete` helper | +| `namespace/configurator/{primary,schema,fork}.rs` | use `FreshDir`; implement `discard_incomplete`; drop ad-hoc cleanup; fork marks its temp dir | +| `namespace/meta_store.rs` | skip marked dirs in `maybe_recover_from_fs` | +| `docs/ADMIN_API.md`, PR #51 design doc §13 | contract; mark the lifecycle item as fixed | +| tests | §11 | + +Branch `tszymczyszyn/atomic-namespace-create` off `tszymczyszyn/streaming-dump-importer`; the PR targets the importer branch until #51 merges, then retargets `v0.9.30-shopify-patches`. + +## 14. Alternatives considered + +- **Row first with a lifecycle state (`creating`/`ready`).** Needs a metastore schema or protobuf change that replicates to replicas via the config handshake, plus a startup sweep that knows configurator paths. More moving parts for the same guarantee. +- **Import into a temp directory and rename.** The logger, stats and connection maker capture the path at open; publishing would require close → rename → reopen, and bottomless would see a restore decision on reopen. `fork` can do it because it only writes a raw `data` file before opening. +- **Dedup via moka `try_get_with` instead of an explicit lock.** When the initialising future is dropped, moka lets another waiter run *its* init — a lazy `with()` would then create an empty namespace. The explicit reservation makes waiters observe the outcome instead. +- **Making `with()` fail fast ("namespace is being created") instead of waiting.** Cleaner UX for minute-long imports, but changes behaviour for `reset`/`fork` too; can be layered on later without touching this design. diff --git a/docs/STREAMING_DUMP_IMPORT_DESIGN.md b/docs/STREAMING_DUMP_IMPORT_DESIGN.md index 525f947071..f8358fffe8 100644 --- a/docs/STREAMING_DUMP_IMPORT_DESIGN.md +++ b/docs/STREAMING_DUMP_IMPORT_DESIGN.md @@ -585,12 +585,12 @@ Metrics (`metrics` 0.21 macros with labels): ## 13. Pre-existing issues this design does not fix (state them in the PR) -1. `NamespaceStore::create` stores the namespace config in the metastore **before** loading. On import failure the directory is removed (`PrimaryConfigurator::setup`) but the metastore row stays, so a later request to the namespace lazily creates an **empty** database. Existing tests rely on this (`select … from test` errors because the table is missing, not because the namespace is missing). -2. If the Admin HTTP request is cancelled mid-import, `setup`'s cleanup does not run (the future is dropped). The streaming executor still rolls back SQLite state, but the directory and `.sentinel` remain. -3. No server-side timeout exists for dump imports; the Admin HTTP request stays open for the whole import. Clients must not time out, or must tolerate (2). +1. ~~`NamespaceStore::create` stores the namespace config in the metastore **before** loading. On import failure the directory is removed (`PrimaryConfigurator::setup`) but the metastore row stays, so a later request to the namespace lazily creates an **empty** database.~~ **Fixed** by the stacked change described in `docs/ATOMIC_NAMESPACE_CREATE_DESIGN.md`: the config is persisted only after setup succeeds, and a failed or cancelled creation leaves nothing behind. +2. ~~If the Admin HTTP request is cancelled mid-import, `setup`'s cleanup does not run (the future is dropped). The streaming executor still rolls back SQLite state, but the directory and `.sentinel` remain.~~ **Fixed**, same change (drop guards for the directory and the reservation of the name). +3. No server-side timeout exists for dump imports; the Admin HTTP request stays open for the whole import. Clients must not time out; if they do, the creation is discarded and can be retried (see the creation contract in `docs/ADMIN_API.md`). 4. Bottomless, if enabled, has its own frame buffering/backpressure outside this design. -These are tracked by the DB Mover quarantine/fence work (Retail #35846/#35848). +(3) and (4) are tracked by the DB Mover quarantine/fence work (Retail #35846/#35848). --- diff --git a/docs/images/atomic-namespace-create-before-after.png b/docs/images/atomic-namespace-create-before-after.png new file mode 100644 index 0000000000..af6172c406 Binary files /dev/null and b/docs/images/atomic-namespace-create-before-after.png differ diff --git a/docs/images/atomic-namespace-create-components.png b/docs/images/atomic-namespace-create-components.png new file mode 100644 index 0000000000..522f26d529 Binary files /dev/null and b/docs/images/atomic-namespace-create-components.png differ diff --git a/libsql-server/Cargo.toml b/libsql-server/Cargo.toml index e9721a32ba..6c11402828 100644 --- a/libsql-server/Cargo.toml +++ b/libsql-server/Cargo.toml @@ -11,7 +11,7 @@ path = "src/main.rs" [dependencies] anyhow = "1.0.66" -async-lock = "2.6.0" +async-lock = "3" async-stream = "0.3.5" async-tempfile = "0.4.0" async-trait = "0.1.58" diff --git a/libsql-server/src/lib.rs b/libsql-server/src/lib.rs index 10b46ae2d3..731f89d3b0 100644 --- a/libsql-server/src/lib.rs +++ b/libsql-server/src/lib.rs @@ -638,6 +638,13 @@ where ) .await?; + // Namespace creations the previous process died in the middle of leave marked + // directories (and tombstones of discarded ones) behind; clean them up before serving. + if let Err(e) = namespace::incomplete::sweep(&self.path, namespace_store.meta_store()).await + { + tracing::error!("failed to sweep unfinished namespace directories: {e}"); + } + self.spawn_monitoring_tasks(&mut task_manager, stats_receiver)?; // if namespaces are enabled, then bottomless must have set DB ID diff --git a/libsql-server/src/namespace/configurator/fork.rs b/libsql-server/src/namespace/configurator/fork.rs index 80de9ab19c..ab82360e25 100644 --- a/libsql-server/src/namespace/configurator/fork.rs +++ b/libsql-server/src/namespace/configurator/fork.rs @@ -130,9 +130,13 @@ impl ForkTask { let dest_namespace = self.to_namespace.clone(); match self.try_fork().await { Err(e) => { - let _ = - tokio::fs::remove_dir_all(base_path.join("dbs").join(dest_namespace.as_str())) - .await; + // Only what this fork put there (the renamed, marked temp dir); a directory that + // was already at the destination is not ours to remove. + if let Err(e) = + crate::namespace::incomplete::discard_if_incomplete(&base_path, &dest_namespace) + { + tracing::error!(namespace = %dest_namespace, "failed to discard fork directory: {e}"); + } Err(e) } Ok(ns) => Ok(ns), @@ -145,6 +149,9 @@ impl ForkTask { let temp_dir = BLOCKING_RT .spawn_blocking(move || tempfile::tempdir_in(base_path)) .await??; + // Marked incomplete until the store has persisted the namespace's config, so that a + // fork that does not complete is discarded like any other unfinished creation. + crate::namespace::incomplete::mark_incomplete(temp_dir.path())?; let db_path = temp_dir.path().join("data"); if let Some(restore) = self.restore_to { diff --git a/libsql-server/src/namespace/configurator/helpers.rs b/libsql-server/src/namespace/configurator/helpers.rs index 023333be30..670f4a727c 100644 --- a/libsql-server/src/namespace/configurator/helpers.rs +++ b/libsql-server/src/namespace/configurator/helpers.rs @@ -19,8 +19,8 @@ use crate::namespace::broadcasters::BroadcasterHandle; use crate::namespace::meta_store::MetaStoreHandle; use crate::namespace::replication_wal::{make_replication_wal_wrapper, ReplicationWalWrapper}; use crate::namespace::{ - NamespaceBottomlessDbId, NamespaceBottomlessDbIdInit, NamespaceName, ResolveNamespacePathFn, - RestoreOption, + incomplete, NamespaceBottomlessDbId, NamespaceBottomlessDbIdInit, NamespaceName, + ResolveNamespacePathFn, RestoreOption, }; use crate::replication::ReplicationLogger; use crate::stats::Stats; @@ -28,6 +28,59 @@ use crate::{StatsSender, BLOCKING_RT, DB_CREATE_TIMEOUT, DEFAULT_AUTO_CHECKPOINT use super::{BaseNamespaceConfig, PrimaryConfig}; +/// The directory of a namespace that does not exist yet, created marked incomplete. +/// +/// If the guard is dropped before [`keep`](Self::keep) — setup failed, or the future creating +/// the namespace was dropped — the directory is discarded (atomically renamed away and deleted +/// in the background, see [`incomplete::discard_dir`]). The marker itself is removed by the +/// namespace store once the config is persisted. +pub(super) struct FreshDir { + base: Arc, + path: Option>, +} + +impl FreshDir { + /// Returns `None` when a namespace directory already exists at `db_path`. + /// + /// Synchronous on purpose: the guard exists before anything is created, and there is no + /// await point between creating the directory and marking it. + pub(super) fn begin(base: &Arc, db_path: &Arc) -> crate::Result> { + if db_path.try_exists()? { + return Ok(None); + } + let guard = Self { + base: base.clone(), + path: Some(db_path.clone()), + }; + incomplete::create_marked_dir(db_path)?; + Ok(Some(guard)) + } + + /// The namespace was set up; the directory stays. + pub(super) fn keep(mut self) { + self.path.take(); + } +} + +impl Drop for FreshDir { + fn drop(&mut self) { + let Some(path) = self.path.take() else { + return; + }; + tracing::info!( + path = %path.display(), + "namespace creation did not complete; discarding its directory" + ); + if let Err(e) = incomplete::discard_dir(&self.base, &path) { + // still marked: discarded by the next creation of the name, or by the startup sweep + tracing::error!( + path = %path.display(), + "failed to discard unfinished namespace directory: {e}" + ); + } + } +} + #[tracing::instrument(skip_all)] pub(super) async fn make_primary_connection_maker( primary_config: &PrimaryConfig, @@ -405,3 +458,90 @@ pub(super) async fn cleanup_primary( Ok(()) } + +#[cfg(test)] +mod test { + use super::*; + use crate::namespace::incomplete::INCOMPLETE_MARKER; + + async fn settle_background_deletes() { + for _ in 0..50 { + tokio::task::yield_now().await; + tokio::time::sleep(Duration::from_millis(2)).await; + } + } + + #[tokio::test] + async fn fresh_dir_is_discarded_unless_kept() { + let tmp = tempfile::tempdir().unwrap(); + let base: Arc = tmp.path().into(); + let path: Arc = tmp.path().join("dbs").join("ns").into(); + + let fresh = FreshDir::begin(&base, &path).unwrap().expect("fresh"); + assert!(path.join(INCOMPLETE_MARKER).exists()); + drop(fresh); + // the path is free synchronously, the tombstone goes away in the background + assert!(!path.exists(), "dropped before keep: discarded"); + settle_background_deletes().await; + assert!(std::fs::read_dir(tmp.path().join(".discarded")) + .unwrap() + .next() + .is_none()); + + let fresh = FreshDir::begin(&base, &path).unwrap().expect("fresh again"); + fresh.keep(); + assert!( + path.join(INCOMPLETE_MARKER).exists(), + "kept: stays, marker included" + ); + + assert!( + FreshDir::begin(&base, &path).unwrap().is_none(), + "an existing directory is not fresh" + ); + } + + #[tokio::test] + async fn fresh_dir_is_discarded_when_its_future_is_dropped() { + let tmp = tempfile::tempdir().unwrap(); + let base: Arc = tmp.path().into(); + let path: Arc = tmp.path().join("dbs").join("ns").into(); + let (ready, is_ready) = tokio::sync::oneshot::channel(); + + let setup = { + let (base, path) = (base.clone(), path.clone()); + async move { + let fresh = FreshDir::begin(&base, &path).unwrap(); + ready.send(()).unwrap(); + std::future::pending::<()>().await; // "the import" + fresh.map(FreshDir::keep); + } + }; + let setup = tokio::spawn(setup); + is_ready.await.unwrap(); + assert!(path.join(INCOMPLETE_MARKER).exists()); + setup.abort(); + let _ = setup.await; + assert!(!path.exists()); + } + + #[tokio::test] + async fn discard_if_incomplete_only_removes_marked_directories() { + let tmp = tempfile::tempdir().unwrap(); + let marked = tmp.path().join("dbs").join("marked"); + let unmarked = tmp.path().join("dbs").join("unmarked"); + for dir in [&marked, &unmarked] { + std::fs::create_dir_all(dir).unwrap(); + std::fs::write(dir.join("data"), b"junk").unwrap(); + } + std::fs::write(marked.join(INCOMPLETE_MARKER), b"").unwrap(); + + let ns = |s: &str| NamespaceName::from_string(s.into()).unwrap(); + assert!(incomplete::discard_if_incomplete(tmp.path(), &ns("marked")).unwrap()); + assert!(!incomplete::discard_if_incomplete(tmp.path(), &ns("unmarked")).unwrap()); + assert!(!incomplete::discard_if_incomplete(tmp.path(), &ns("absent")).unwrap()); + + assert!(!marked.exists()); + assert!(unmarked.join("data").exists()); + } +} diff --git a/libsql-server/src/namespace/configurator/mod.rs b/libsql-server/src/namespace/configurator/mod.rs index 064f9a8f7c..5e5c43cd9f 100644 --- a/libsql-server/src/namespace/configurator/mod.rs +++ b/libsql-server/src/namespace/configurator/mod.rs @@ -131,6 +131,16 @@ pub trait ConfigureNamespace { bottomless_db_id_init: NamespaceBottomlessDbIdInit, ) -> Pin> + Send + 'a>>; + /// Discard whatever a creation of `namespace` that did not complete left on disk, so that + /// the name starts from nothing. Only called for a name that is neither loaded nor known to + /// the metastore; must leave anything that does not carry the + /// [`INCOMPLETE_MARKER`](super::incomplete::INCOMPLETE_MARKER) alone. Synchronous so that + /// it can run while a creating future is being dropped (see `NamespaceStore`). + fn discard_incomplete(&self, namespace: &NamespaceName) -> crate::Result<()> { + let _ = namespace; + Ok(()) + } + fn fork<'a>( &'a self, from_ns: &'a Namespace, diff --git a/libsql-server/src/namespace/configurator/primary.rs b/libsql-server/src/namespace/configurator/primary.rs index f68405fad6..13297d0144 100644 --- a/libsql-server/src/namespace/configurator/primary.rs +++ b/libsql-server/src/namespace/configurator/primary.rs @@ -21,7 +21,7 @@ use crate::namespace::{ use crate::run_periodic_checkpoint; use crate::schema::{has_pending_migration_task, setup_migration_table}; -use super::helpers::cleanup_primary; +use super::helpers::{cleanup_primary, FreshDir}; use super::{BaseNamespaceConfig, ConfigureNamespace, PrimaryConfig}; pub struct PrimaryConfigurator { @@ -129,35 +129,31 @@ impl ConfigureNamespace for PrimaryConfigurator { ) -> Pin> + Send + 'a>> { Box::pin(async move { let db_path: Arc = self.base.base_path.join("dbs").join(name.as_str()).into(); - let fresh_namespace = !db_path.try_exists()?; - // FIXME: make that truly atomic. explore the idea of using temp directories, and it's implications - match self + // A brand-new directory is discarded again if setup fails or this future is dropped. + let fresh = FreshDir::begin(&self.base.base_path, &db_path)?; + let ns = self .try_new_primary( name.clone(), meta_store_handle, restore_option, resolve_attach_path, - db_path.clone(), + db_path, broadcaster, self.base.encryption_config.clone(), ) - .await - { - Ok(this) => Ok(this), - Err(e) if fresh_namespace => { - tracing::error!( - "an error occured while deleting creating namespace, cleaning..." - ); - if let Err(e) = tokio::fs::remove_dir_all(&db_path).await { - tracing::error!("failed to remove dirty namespace directory: {e}") - } - Err(e) - } - Err(e) => Err(e), + .await?; + if let Some(fresh) = fresh { + fresh.keep(); } + Ok(ns) }) } + fn discard_incomplete(&self, namespace: &NamespaceName) -> crate::Result<()> { + crate::namespace::incomplete::discard_if_incomplete(&self.base.base_path, namespace)?; + Ok(()) + } + fn cleanup<'a>( &'a self, namespace: &'a NamespaceName, diff --git a/libsql-server/src/namespace/configurator/schema.rs b/libsql-server/src/namespace/configurator/schema.rs index 275fd71e93..956b0141b1 100644 --- a/libsql-server/src/namespace/configurator/schema.rs +++ b/libsql-server/src/namespace/configurator/schema.rs @@ -13,7 +13,7 @@ use crate::namespace::{ }; use crate::schema::SchedulerHandle; -use super::helpers::{cleanup_primary, make_primary_connection_maker}; +use super::helpers::{cleanup_primary, make_primary_connection_maker, FreshDir}; use super::{BaseNamespaceConfig, ConfigureNamespace, PrimaryConfig}; pub struct SchemaConfigurator { @@ -52,9 +52,10 @@ impl ConfigureNamespace for SchemaConfigurator { ) -> std::pin::Pin> + Send + 'a>> { Box::pin(async move { let mut join_set = JoinSet::new(); - let db_path = self.base.base_path.join("dbs").join(name.as_str()); - - tokio::fs::create_dir_all(&db_path).await?; + let db_path: Arc = + self.base.base_path.join("dbs").join(name.as_str()).into(); + // A brand-new directory is discarded again if setup fails or this future is dropped. + let fresh = FreshDir::begin(&self.base.base_path, &db_path)?; let (connection_maker, wal_manager, stats) = make_primary_connection_maker( &self.primary_config, @@ -72,6 +73,10 @@ impl ConfigureNamespace for SchemaConfigurator { ) .await?; + if let Some(fresh) = fresh { + fresh.keep(); + } + Ok(Namespace { db: Database::Schema(SchemaDatabase::new( self.migration_scheduler.clone(), @@ -89,11 +94,16 @@ impl ConfigureNamespace for SchemaConfigurator { tasks: join_set, stats, db_config_store: db_config.clone(), - path: db_path.into(), + path: db_path, }) }) } + fn discard_incomplete(&self, namespace: &NamespaceName) -> crate::Result<()> { + crate::namespace::incomplete::discard_if_incomplete(&self.base.base_path, namespace)?; + Ok(()) + } + fn cleanup<'a>( &'a self, namespace: &'a NamespaceName, diff --git a/libsql-server/src/namespace/incomplete.rs b/libsql-server/src/namespace/incomplete.rs new file mode 100644 index 0000000000..3c907ec059 --- /dev/null +++ b/libsql-server/src/namespace/incomplete.rs @@ -0,0 +1,270 @@ +//! On-disk protocol for namespaces whose creation has not completed. +//! +//! A namespace directory carries [`INCOMPLETE_MARKER`] from the moment it is created until the +//! namespace's config has been persisted to the metastore. A directory that still carries it +//! belongs to a creation that did not complete — the request failed or was cancelled and the +//! cleanup could not finish, or the process died — and is garbage. +//! +//! Garbage is never deleted in place: [`discard_dir`] atomically renames the directory into +//! `/.discarded/` first, so that its path under `dbs/` is free immediately and nothing that +//! still has files open in it (an import executor finishing its rollback, say) can recreate +//! anything under the old path, and deletes the tombstone in the background. [`sweep`] removes +//! tombstones and marked orphans left by a crash at startup. +//! +//! See `docs/ATOMIC_NAMESPACE_CREATE_DESIGN.md`. + +use std::path::{Path, PathBuf}; +#[cfg(test)] +use std::sync::Arc; +#[cfg(test)] +use std::time::Duration; + +#[cfg(test)] +use crate::connection::config::DatabaseConfig; + +use super::meta_store::MetaStore; +use super::NamespaceName; + +/// Present in a namespace directory until its config has been persisted to the metastore. +pub(crate) const INCOMPLETE_MARKER: &str = ".incomplete"; + +/// Where discarded directories go before they are deleted. +const DISCARDED_DIR: &str = ".discarded"; + +/// Path of the directory of namespace `name`. +pub(crate) fn namespace_dir(base: &Path, name: &NamespaceName) -> PathBuf { + base.join("dbs").join(name.as_str()) +} + +/// Create the directory of a namespace that does not exist yet, marked incomplete. +/// +/// Synchronous (a few small filesystem operations) so that there is no await point between +/// creating the directory and marking it. The marker and the directory entry are fsynced: the +/// marker must be durable before anything else in the directory can be, or a power loss could +/// leave a directory with content but no marker, which nothing would dare discard. +pub(crate) fn create_marked_dir(db_path: &Path) -> std::io::Result<()> { + std::fs::create_dir_all(db_path)?; + mark_incomplete(db_path) +} + +/// Write the marker into an existing directory and make it durable. +pub(crate) fn mark_incomplete(db_path: &Path) -> std::io::Result<()> { + std::fs::File::create(db_path.join(INCOMPLETE_MARKER))?.sync_all()?; + std::fs::File::open(db_path)?.sync_all() +} + +/// Remove `db_path` (a namespace directory under `base`): rename it into the discarded area +/// right away and delete it in the background. Returns `Ok` if it does not exist. +/// +/// Synchronous so that it can run in `Drop` and on the blocking pool; the rename is instant, +/// the deletion is handed to the runtime (or done inline if there is none). +pub(crate) fn discard_dir(base: &Path, db_path: &Path) -> std::io::Result<()> { + let discarded = base.join(DISCARDED_DIR); + std::fs::create_dir_all(&discarded)?; + let name = db_path + .file_name() + .map(|n| n.to_string_lossy().into_owned()) + .unwrap_or_default(); + let tombstone = discarded.join(format!("{name}-{}", uuid::Uuid::new_v4())); + match std::fs::rename(db_path, &tombstone) { + Ok(()) => {} + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(e) => return Err(e), + } + tracing::info!( + path = %db_path.display(), + tombstone = %tombstone.display(), + "discarded namespace directory" + ); + delete_in_background(tombstone); + Ok(()) +} + +/// Discard the directory of `name` under `base` if it carries the marker. Directories without +/// the marker are left alone. +pub(crate) fn discard_if_incomplete(base: &Path, name: &NamespaceName) -> std::io::Result { + let db_path = namespace_dir(base, name); + if !db_path.join(INCOMPLETE_MARKER).try_exists()? { + return Ok(false); + } + tracing::warn!( + namespace = %name, + path = %db_path.display(), + "discarding namespace directory left by a creation that did not complete" + ); + discard_dir(base, &db_path)?; + metrics::increment_counter!("libsql_server_namespace_create_discarded_incomplete"); + Ok(true) +} + +fn delete_in_background(tombstone: PathBuf) { + let delete = move || { + if let Err(e) = std::fs::remove_dir_all(&tombstone) { + if e.kind() != std::io::ErrorKind::NotFound { + // left for `sweep` at the next start + tracing::error!( + path = %tombstone.display(), + "failed to delete discarded namespace directory: {e}" + ); + } + } + }; + match tokio::runtime::Handle::try_current() { + Ok(rt) => { + rt.spawn_blocking(delete); + } + Err(_) => delete(), + } +} + +/// Startup sweep: delete leftover tombstones and discard marked directories of namespaces the +/// metastore does not know (creations the process died in the middle of). +pub(crate) async fn sweep(base: &Path, meta_store: &MetaStore) -> std::io::Result<()> { + let discarded = base.join(DISCARDED_DIR); + if let Ok(entries) = std::fs::read_dir(&discarded) { + for entry in entries.flatten() { + tracing::warn!(path = %entry.path().display(), "deleting leftover discarded namespace directory"); + delete_in_background(entry.path()); + } + } + + let dbs = base.join("dbs"); + let Ok(entries) = std::fs::read_dir(&dbs) else { + return Ok(()); + }; + for entry in entries.flatten() { + if !entry.path().join(INCOMPLETE_MARKER).try_exists()? { + continue; + } + let Ok(name) = NamespaceName::from_string(entry.file_name().to_string_lossy().into_owned()) + else { + continue; + }; + if meta_store.exists(&name).await { + // persisted, but the marker was not removed yet (crash right after the commit); it + // is removed when the namespace is loaded + continue; + } + discard_if_incomplete(base, &name)?; + } + Ok(()) +} + +#[cfg(test)] +mod test { + use super::*; + use crate::database::DatabaseKind; + use crate::namespace::meta_store::metastore_connection_maker; + + async fn meta_store(path: &Path) -> MetaStore { + let (maker, manager) = metastore_connection_maker(None, path).await.unwrap(); + MetaStore::new( + Default::default(), + path, + maker().unwrap(), + manager, + DatabaseKind::Primary, + ) + .await + .unwrap() + } + + fn ns(s: &str) -> NamespaceName { + NamespaceName::from_string(s.into()).unwrap() + } + + #[tokio::test] + async fn discard_dir_frees_the_path_immediately() { + let tmp = tempfile::tempdir().unwrap(); + let dir = tmp.path().join("dbs").join("foo"); + create_marked_dir(&dir).unwrap(); + std::fs::write(dir.join("data"), b"junk").unwrap(); + // a file is still open inside; the rename doesn't care, and neither does the deletion + let _open = std::fs::File::open(dir.join("data")).unwrap(); + + discard_dir(tmp.path(), &dir).unwrap(); + assert!(!dir.exists()); + // idempotent + discard_dir(tmp.path(), &dir).unwrap(); + + for _ in 0..100 { + if std::fs::read_dir(tmp.path().join(DISCARDED_DIR)) + .unwrap() + .next() + .is_none() + { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("tombstone was not deleted"); + } + + /// Startup: tombstones are deleted; marked directories of names the metastore doesn't + /// know are discarded; marked directories of known names and unmarked directories stay. + #[tokio::test] + async fn sweep_cleans_leftovers_of_a_crash() { + let tmp = tempfile::tempdir().unwrap(); + let base = tmp.path(); + let store = meta_store(base).await; + store + .handle(ns("known")) + .await + .store(Arc::new(DatabaseConfig::default())) + .await + .unwrap(); + + let tombstone = base.join(DISCARDED_DIR).join("old-123"); + std::fs::create_dir_all(&tombstone).unwrap(); + std::fs::write(tombstone.join("data"), b"junk").unwrap(); + let crashed = base.join("dbs").join("crashed"); + create_marked_dir(&crashed).unwrap(); + let known = base.join("dbs").join("known"); + create_marked_dir(&known).unwrap(); + let unmarked = base.join("dbs").join("unmarked"); + std::fs::create_dir_all(&unmarked).unwrap(); + + sweep(base, &store).await.unwrap(); + + assert!(!crashed.exists()); + assert!( + known.join(INCOMPLETE_MARKER).exists(), + "known: left for load" + ); + assert!(unmarked.exists()); + for _ in 0..100 { + if !tombstone.exists() { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("tombstone was not deleted"); + } + + /// Rebuilding an empty metastore from the filesystem must not adopt a directory whose + /// creation never completed. + #[tokio::test] + async fn recovery_from_fs_skips_marked_directories() { + let tmp = tempfile::tempdir().unwrap(); + let base = tmp.path(); + std::fs::create_dir_all(base.join("dbs").join("complete")).unwrap(); + create_marked_dir(&base.join("dbs").join("crashed")).unwrap(); + + let (maker, manager) = metastore_connection_maker(None, base).await.unwrap(); + let store = MetaStore::new( + crate::config::MetaStoreConfig { + allow_recover_from_fs: true, + ..Default::default() + }, + base, + maker().unwrap(), + manager, + DatabaseKind::Primary, + ) + .await + .unwrap(); + + assert!(store.exists(&ns("complete")).await); + assert!(!store.exists(&ns("crashed")).await); + } +} diff --git a/libsql-server/src/namespace/meta_store.rs b/libsql-server/src/namespace/meta_store.rs index 70b419ebe9..1f984c49b6 100644 --- a/libsql-server/src/namespace/meta_store.rs +++ b/libsql-server/src/namespace/meta_store.rs @@ -30,12 +30,23 @@ use crate::{ use super::NamespaceName; -type ChangeMsg = ( - NamespaceName, - Option>, - oneshot::Sender>, - bool, // flush -); +enum ChangeMsg { + /// Set the in-memory config of a namespace (`config: Some`) and/or persist its current + /// in-memory config (`flush`). + Update { + namespace: NamespaceName, + config: Option>, + ret: oneshot::Sender>, + flush: bool, + }, + /// Forget a namespace whose creation did not complete: drop its in-memory config and + /// whatever was persisted for it, unconditionally. Goes through the same queue as updates + /// so that it is ordered after any update to the same namespace that is still queued. + Discard { + namespace: NamespaceName, + ret: oneshot::Sender>, + }, +} type MetaStoreWalManager = WalWrapper, Sqlite3WalManager>; pub type MetaStoreConnection = libsql_sys::Connection, Sqlite3Wal>>; @@ -223,6 +234,17 @@ impl MetaStoreInner { if !entry.path().is_dir() { continue; } + if entry + .path() + .join(super::incomplete::INCOMPLETE_MARKER) + .try_exists()? + { + tracing::warn!( + "skipping `{}`: left by an interrupted namespace creation", + entry.path().display() + ); + continue; + } let config_path = entry.path().join("config.json"); let name = NamespaceName::from_string(entry.file_name().to_str().unwrap().to_string())?; @@ -306,7 +328,18 @@ impl MetaStoreInner { /// Handles config change updates by inserting them into the database and in-memory /// cache of configs. fn process(msg: ChangeMsg, inner: Arc) { - let (namespace, config, ret_chan, flush) = msg; + let (namespace, config, ret_chan, flush) = match msg { + ChangeMsg::Update { + namespace, + config, + ret, + flush, + } => (namespace, config, ret, flush), + ChangeMsg::Discard { namespace, ret } => { + let _ = ret.send(discard(&inner, &namespace)); + return; + } + }; if let Some(config) = config { let ret = if flush { try_process(&inner, &namespace, &config) @@ -330,12 +363,16 @@ fn process(msg: ChangeMsg, inner: Arc) { let _ = ret_chan.send(ret); } else { let ret = if flush { - let mut configs = inner.configs.blocking_lock(); - if let Some(config_watch) = configs.get_mut(&namespace) { - let config = config_watch.subscribe().borrow().clone(); - try_process(&inner, &namespace, &config.config) - } else { - Ok(()) + // Lock order: `remove` takes `conn` before `configs`; `try_process` takes `conn`, + // so `configs` must not be held across it. + let current = inner + .configs + .blocking_lock() + .get(&namespace) + .map(|config_watch| config_watch.borrow().config.clone()); + match current { + Some(config) => try_process(&inner, &namespace, &config), + None => Ok(()), } } else { Ok(()) @@ -344,6 +381,27 @@ fn process(msg: ChangeMsg, inner: Arc) { } } +/// Forget `namespace` unconditionally: the row and schema links, if any were written, and the +/// in-memory config. No consistency checks (unlike [`MetaStore::remove`]): this undoes a +/// creation that did not complete, so nothing can depend on the namespace yet. +fn discard(inner: &MetaStoreInner, namespace: &NamespaceName) -> Result<()> { + { + let mut conn = inner.conn.blocking_lock(); + let tx = conn.transaction()?; + tx.execute( + "DELETE FROM shared_schema_links WHERE namespace = ?", + [namespace.as_str()], + )?; + tx.execute( + "DELETE FROM namespace_configs WHERE namespace = ?", + [namespace.as_str()], + )?; + tx.commit()?; + } + inner.configs.blocking_lock().remove(namespace); + Ok(()) +} + fn try_process( inner: &MetaStoreInner, namespace: &NamespaceName, @@ -508,6 +566,31 @@ impl MetaStore { } } + /// Handle of a namespace the store knows; `None` otherwise. Unlike [`handle`](Self::handle) + /// this never registers the namespace, so it is the lookup for anything that must not + /// bring a namespace into existence as a side effect. + pub async fn existing_handle(&self, namespace: NamespaceName) -> Option { + let configs = self.inner.configs.lock().await; + let rx = configs.get(&namespace)?.subscribe(); + Some(MetaStoreHandle { + namespace, + inner: HandleState::External(self.changes_tx.clone(), rx), + }) + } + + /// Forget a namespace whose creation did not complete (see [`ChangeMsg::Discard`]). + /// + /// Blocking: meant for the blocking pool (the cleanup runs where the creating future is + /// being dropped). + pub fn discard_blocking(&self, namespace: NamespaceName) -> Result<()> { + let (ret, rcv) = oneshot::channel(); + self.changes_tx + .blocking_send(ChangeMsg::Discard { namespace, ret }) + .map_err(|e| Error::MetaStoreUpdateFailure(e.into()))?; + rcv.blocking_recv() + .map_err(|e| Error::MetaStoreUpdateFailure(e.into()))? + } + pub fn remove(&self, namespace: NamespaceName) -> Result>> { tracing::debug!("removing namespace `{}` from meta store", namespace); @@ -708,7 +791,12 @@ impl MetaStoreHandle { let (snd, rcv) = oneshot::channel(); changes_tx - .send((self.namespace.clone(), new_config, snd, flush)) + .send(ChangeMsg::Update { + namespace: self.namespace.clone(), + config: new_config, + ret: snd, + flush, + }) .await .map_err(|e| Error::MetaStoreUpdateFailure(e.into()))?; diff --git a/libsql-server/src/namespace/mod.rs b/libsql-server/src/namespace/mod.rs index 2cea87ed75..5a3b6b0472 100644 --- a/libsql-server/src/namespace/mod.rs +++ b/libsql-server/src/namespace/mod.rs @@ -21,6 +21,7 @@ pub use self::store::NamespaceStore; pub mod broadcasters; pub(crate) mod configurator; pub mod dump_import; +pub(crate) mod incomplete; pub mod meta_store; mod name; pub mod replication_wal; @@ -56,6 +57,20 @@ pub enum NamespaceBottomlessDbIdInit { FetchFromConfig, } +/// Remove the incomplete marker of the namespace directory `path`, if present. +pub(crate) async fn remove_incomplete_marker(path: &Path) -> std::io::Result<()> { + let marker = path.join(incomplete::INCOMPLETE_MARKER); + // Almost always absent (every cold load passes through here): a stat is cheaper than a + // trip to the blocking pool. + if !marker.exists() { + return Ok(()); + } + match tokio::fs::remove_file(marker).await { + Err(e) if e.kind() != std::io::ErrorKind::NotFound => Err(e), + _ => Ok(()), + } +} + /// A namespace isolates the resources pertaining to a database of type T #[derive(Debug)] pub struct Namespace { @@ -73,6 +88,12 @@ impl Namespace { &self.name } + /// Remove the [`incomplete::INCOMPLETE_MARKER`]: the namespace is now persisted in the + /// metastore (or was opened from it, which implies the same). + pub(crate) async fn mark_complete(&self) -> std::io::Result<()> { + remove_incomplete_marker(&self.path).await + } + async fn destroy(mut self) -> anyhow::Result<()> { self.tasks.shutdown().await; self.db.destroy(); diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index 86e9438ccd..69ed1138f6 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -1,11 +1,15 @@ +use std::collections::HashMap; +use std::future::Future; +use std::pin::Pin; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; -use async_lock::RwLock; +use async_lock::{RwLock, RwLockWriteGuardArc}; use chrono::NaiveDateTime; use futures::TryFutureExt; use moka::future::Cache; use once_cell::sync::OnceCell; +use parking_lot::Mutex; use tokio::task::JoinSet; use tokio::time::{Duration, Instant}; use tokio_stream::wrappers::BroadcastStream; @@ -27,6 +31,205 @@ use super::{Namespace, ResetCb, ResetOp, ResolveNamespacePathFn, RestoreOption}; type NamespaceEntry = Arc>>; +/// Entries of namespaces with a creation, fork or reset in progress, pinned so that lookups keep +/// finding (and waiting on) the locked entry even if the cache evicts it meanwhile. Otherwise a +/// concurrent lookup would run a lazy load of the directory the operation is writing to. +type InFlight = Arc>>; + +/// Remove `name` from `in_flight`, but only if it still maps to `entry` (another operation may +/// have taken the name over in the meantime). +fn unpin(in_flight: &InFlight, name: &NamespaceName, entry: &NamespaceEntry) { + let mut in_flight = in_flight.lock(); + if in_flight.get(name).is_some_and(|e| Arc::ptr_eq(e, entry)) { + in_flight.remove(name); + } +} + +/// A namespace name reserved for a creation (or fork) in progress. +/// +/// Holding a reservation holds the name's cache entry write-locked, so every other access to +/// the name — user requests, a second create, destroy — waits until the outcome is known. +/// [`publish`](Self::publish) installs the finished namespace. Dropping the reservation +/// unpublished (error, or the creating request was cancelled) undoes everything the creation +/// did — the namespace's config, and its directory if it is still marked incomplete — and only +/// then releases the lock, so waiters observe a namespace that does not exist, never a +/// half-built one, and a retry never races the cleanup. +/// +/// The entry is pinned in [`InFlight`] for the duration, so that lookups keep finding it (and +/// waiting on it) even if the cache evicts it in the meantime. +/// +/// The metastore row is only written after setup succeeds (see `create`), so the row's presence +/// implies a complete namespace. See `docs/ATOMIC_NAMESPACE_CREATE_DESIGN.md`. +struct Reservation { + guard: Option>>, + entry: NamespaceEntry, + store: NamespaceStore, + name: NamespaceName, + /// Selects the configurator that owns the namespace's directory. + config: Arc, +} + +impl Reservation { + /// Fails with `NamespaceAlreadyExist` if the namespace is loaded or known to the metastore. + /// + /// Must be called before `MetaStore::handle(name)`: `handle` inserts the name into the + /// in-memory config map that `exists` reads, so the name becomes visible only while it is + /// already locked here. + async fn acquire( + store: &NamespaceStore, + name: NamespaceName, + config: Arc, + ) -> crate::Result { + // Wait on the entry an operation in progress holds, if any, rather than on the cache's + // (which may be a fresh one if that entry was evicted). + let pinned = store.inner.in_flight.lock().get(&name).cloned(); + let entry = match pinned { + Some(entry) => entry, + None => { + store + .inner + .store + .get_with(name.clone(), async { Default::default() }) + .await + } + }; + if entry.try_write().is_none() { + tracing::debug!(namespace = %name, "waiting for an operation in progress on the namespace"); + } + let guard = entry.write_arc().await; + if guard.is_some() || store.inner.metadata.exists(&name).await { + if guard.is_none() { + // The namespace exists but is not loaded, and `get_with` just gave it an empty + // entry. Left there, lookups would find the empty entry instead of loading the + // namespace (see `load_namespace`); remove it again. + drop(guard); + store.inner.store.invalidate(&name).await; + } + return Err(Error::NamespaceAlreadyExist(name.to_string())); + } + store + .inner + .in_flight + .lock() + .insert(name.clone(), entry.clone()); + Ok(Self { + guard: Some(guard), + entry, + store: store.clone(), + name, + config, + }) + } + + /// Install the finished namespace and release the lock. + async fn publish(mut self, ns: Namespace) { + // `install` awaits (cache lookups). Keep the guard inside `self` across them: if this + // future is dropped there, `Drop` still sees an unpublished reservation and undoes the + // creation — rather than leaving the name pinned and the lock released. + { + let guard = self + .guard + .as_deref_mut() + .expect("a reservation is published at most once"); + self.store.install(&self.name, &self.entry, guard, ns).await; + } + let guard = self.guard.take().expect("checked above"); + // Unpin before unlocking: a lookup racing with us either still finds the pinned entry + // (and waits for the lock, then sees the namespace) or goes through the cache, where + // the namespace now is. + unpin(&self.store.inner.in_flight, &self.name, &self.entry); + drop(guard); + } +} + +impl Reservation { + /// The creation failed: undo it, then release the name. Awaited, so that by the time the + /// error reaches the caller the name is free and nothing is left of the attempt. + async fn abort(mut self) { + if let Some(cleanup) = self.take_cleanup() { + let _ = tokio::task::spawn_blocking(cleanup).await; + } + } + + /// Everything that undoes an unpublished creation, in the order that keeps the name + /// unobservable until it is clean. Blocking (`discard_blocking`). + fn take_cleanup(&mut self) -> Option { + let guard = self.guard.take()?; + metrics::increment_counter!("libsql_server_namespace_create_aborted"); + let store = self.store.clone(); + let name = self.name.clone(); + let entry = self.entry.clone(); + let config = self.config.clone(); + Some(move || { + // 1. The config: in memory and, if the commit happened but publishing didn't, the + // row. From here on lookups of the name are rejected before they reach the cache. + if let Err(e) = store.inner.metadata.discard_blocking(name.clone()) { + tracing::error!(namespace = %name, "failed to discard namespace config: {e}"); + } + // 2. The directory, if it is still marked incomplete (the configurator's own guard + // normally got it already; this covers forks and anything that guard missed). + if let Err(e) = store.get_configurator(&config).discard_incomplete(&name) { + tracing::error!(namespace = %name, "failed to discard namespace directory: {e}"); + } + // 3. Only now let anyone else at the name. + unpin(&store.inner.in_flight, &name, &entry); + drop(guard); + }) + } +} + +impl Drop for Reservation { + /// Neither published nor aborted: the creating future was dropped (the request was + /// cancelled, or the server is shutting down). + fn drop(&mut self) { + let Some(cleanup) = self.take_cleanup() else { + return; + }; + tracing::warn!( + namespace = %self.name, + "namespace creation was cancelled; discarding it" + ); + // We are running inside a future that is being dropped: hand the (blocking) cleanup to + // the blocking pool instead of blocking in place. The name stays locked until it ran. + match tokio::runtime::Handle::try_current() { + Ok(rt) => { + rt.spawn_blocking(cleanup); + } + Err(_) => cleanup(), + } + } +} + +/// Keeps a namespace entry pinned in [`InFlight`] while `reset` holds it locked (see +/// [`Reservation`] for why). Must be created after the lock is taken and dropped before it is +/// released. +struct Pinned { + in_flight: InFlight, + name: NamespaceName, + entry: NamespaceEntry, +} + +impl Pinned { + fn new(store: &NamespaceStore, name: NamespaceName, entry: NamespaceEntry) -> Self { + store + .inner + .in_flight + .lock() + .insert(name.clone(), entry.clone()); + Self { + in_flight: store.inner.in_flight.clone(), + name, + entry, + } + } +} + +impl Drop for Pinned { + fn drop(&mut self) { + unpin(&self.in_flight, &self.name, &self.entry); + } +} + /// Stores and manage a set of namespaces. pub struct NamespaceStore { pub inner: Arc, @@ -50,6 +253,7 @@ pub struct NamespaceStoreInner { broadcasters: BroadcasterRegistry, configurators: NamespaceConfigurators, db_kind: DatabaseKind, + in_flight: InFlight, } impl NamespaceStore { @@ -62,23 +266,41 @@ impl NamespaceStore { db_kind: DatabaseKind, ) -> crate::Result { tracing::trace!("Max active namespaces: {max_active_namespaces}"); + let in_flight: InFlight = Default::default(); let store = Cache::::builder() - .async_eviction_listener(move |name, ns, cause| { - tracing::debug!("evicting namespace `{name}` asynchronously: {cause:?}"); - // TODO(sarna): not clear if we should snapshot-on-evict... - // On the one hand, better to do so, because we have no idea - // for how long we're evicting a namespace. - // On the other, if there's lots of cache pressure, snapshotting - // very often will kill the machine's I/O. - Box::pin(async move { - tracing::info!("namespace `{name}` deallocated"); - // shutdown namespace - if let Some(ns) = ns.write().await.take() { - if let Err(e) = ns.shutdown(snapshot_at_shutdown).await { - tracing::error!("error deallocating `{name}`: {e}") + .async_eviction_listener({ + let in_flight = in_flight.clone(); + move |name, ns, cause| { + tracing::debug!("evicting namespace `{name}` asynchronously: {cause:?}"); + let pinned = in_flight.lock().contains_key(&name); + // TODO(sarna): not clear if we should snapshot-on-evict... + // On the one hand, better to do so, because we have no idea + // for how long we're evicting a namespace. + // On the other, if there's lots of cache pressure, snapshotting + // very often will kill the machine's I/O. + let shutdown = async move { + if let Some(ns) = ns.write().await.take() { + tracing::info!("namespace `{name}` deallocated"); + if let Err(e) = ns.shutdown(snapshot_at_shutdown).await { + tracing::error!("error deallocating `{name}`: {e}") + } } + }; + // moka runs this listener inline, from whichever cache operation happens + // to run maintenance, and awaits it. Normally that is what we want: it + // serialises the shutdown with a reload of the same name. But an entry + // write-locked by an operation in progress (create, fork, reset) stays + // locked for minutes; waiting for it here would stall the cache for + // everyone and deadlock the operation if it touches the cache itself. That + // operation installs its result elsewhere (see `install`), so for it the + // wait only has to happen off the maintenance path. + if pinned { + tokio::spawn(shutdown); + Box::pin(async {}) as Pin + Send>> + } else { + Box::pin(shutdown) } - }) + } }) .max_capacity(max_active_namespaces as u64) .time_to_idle(Duration::from_secs(86400)) @@ -95,6 +317,7 @@ impl NamespaceStore { broadcasters: Default::default(), configurators, db_kind, + in_flight, }), }) } @@ -122,7 +345,12 @@ impl NamespaceStore { .await??; let mut bottomless_db_id_init = NamespaceBottomlessDbIdInit::FetchFromConfig; - if let Some(ns) = self.inner.store.remove(&namespace).await { + // An operation in progress holds the namespace in its pinned entry (the cache's may be + // a fresh, empty one if that entry was evicted): wait on that one, so that we destroy + // what the operation produces rather than racing it. + let pinned = self.inner.in_flight.lock().get(&namespace).cloned(); + let cached = self.inner.store.remove(&namespace).await; + if let Some(ns) = pinned.or(cached) { // deallocate in-memory resources if let Some(ns) = ns.write().await.take() { bottomless_db_id_init = NamespaceBottomlessDbIdInit::Provided( @@ -141,11 +369,11 @@ impl NamespaceStore { } pub async fn checkpoint(&self, namespace: NamespaceName) -> crate::Result<()> { - let entry = self - .inner - .store - .get_with(namespace.clone(), async { Default::default() }) - .await; + // A namespace that is not loaded has nothing to checkpoint. (`get_with(Default)` here + // used to leave an empty entry behind that lookups then mistook for the namespace.) + let Some(entry) = self.inner.store.get(&namespace).await else { + return Ok(()); + }; let lock = entry.read().await; if let Some(ns) = &*lock { ns.checkpoint().await?; @@ -169,6 +397,8 @@ impl NamespaceStore { .get_with(namespace.clone(), async { Default::default() }) .await; let mut lock = entry.write().await; + // Declared after the lock so that it is unpinned before the lock is released. + let _pinned = Pinned::new(self, namespace.clone(), entry.clone()); if let Some(ns) = lock.take() { ns.destroy().await?; } @@ -186,7 +416,7 @@ impl NamespaceStore { .make_namespace(&namespace, db_config, restore_option) .await?; - lock.replace(ns); + self.install(&namespace, &entry, &mut lock, ns).await; Ok(()) } @@ -225,60 +455,42 @@ impl NamespaceStore { return Err(crate::error::Error::NamespaceDoesntExist(from.to_string())); } - let to_entry = self - .inner - .store - .get_with(to.clone(), async { Default::default() }) - .await; - let mut to_lock = to_entry.write().await; - if to_lock.is_some() { - return Err(crate::error::Error::NamespaceAlreadyExist(to.to_string())); + let to_config = Arc::new(to_config); + let reservation = Reservation::acquire(self, to.clone(), to_config.clone()).await?; + match self.fork_reserved(&from, &to, to_config, timestamp).await { + Ok(to_ns) => { + self.publish(reservation, to_ns).await; + Ok(()) + } + Err(e) => { + tracing::warn!(from = %from, to = %to, error = %e, "fork failed"); + reservation.abort().await; + Err(e) + } } + } + /// The part of `fork` that runs with the target name reserved. + async fn fork_reserved( + &self, + from: &NamespaceName, + to: &NamespaceName, + to_config: Arc, + timestamp: Option, + ) -> crate::Result { // FIXME: we could potentially delete the namespace while trying to fork it - if !self.inner.metadata.exists(&from).await { + let Some(from_config) = self.inner.metadata.existing_handle(from.clone()).await else { return Err(crate::Error::NamespaceDoesntExist(from.to_string())); - } - - let from_config = self.inner.metadata.handle(from.clone()).await; - let from_entry = self - .load_namespace(&from, from_config.clone(), RestoreOption::Latest) - .await?; + }; + let from_entry = self.load_namespace(from, from_config.clone()).await?; let from_lock = from_entry.read().await; let Some(from_ns) = &*from_lock else { return Err(crate::error::Error::NamespaceDoesntExist(from.to_string())); }; - struct Bomb { - store: MetaStore, - ns: NamespaceName, - should_delete: bool, - } - - impl Drop for Bomb { - fn drop(&mut self) { - if self.should_delete { - // we need to block in place because the inner connection may blocking, or - // unsing tokio's blocking methods (bottomless), which would cause a panic. - if let Err(e) = - tokio::task::block_in_place(|| self.store.remove(self.ns.clone())) - { - tracing::error!("failed to clean handle while forking: {e}"); - } - } - } - } - - let mut bomb = Bomb { - store: self.inner.metadata.clone(), - ns: to.clone(), - should_delete: true, - }; - let handle = self.inner.metadata.handle(to.clone()).await; - handle - .store_and_maybe_flush(Some(to_config.into()), false) - .await?; + // In memory only: the fork reads it; nothing is persisted until it succeeded. + handle.store_and_maybe_flush(Some(to_config), false).await?; let to_ns = self .get_configurator(&from_config.get()) .fork( @@ -291,12 +503,27 @@ impl NamespaceStore { ) .await?; - to_lock.replace(to_ns); - handle.flush().await?; - // defuse - bomb.should_delete = false; + if self.inner.has_shutdown.load(Ordering::Relaxed) { + return Err(Error::NamespaceStoreShutdown); + } + handle.flush().await?; // commit point + Ok(to_ns) + } - Ok(()) + /// Make a namespace whose config has just been persisted visible, then drop its marker. + /// + /// Visible first, then unmarked: a marker that outlives the namespace's publication is + /// harmless (removed when the namespace is next loaded), a complete directory without one + /// is not (it could not be told from a namespace that was never ours to discard). + async fn publish(&self, reservation: Reservation, ns: Namespace) { + let name = reservation.name.clone(); + let marker_dir = ns.path.clone(); + reservation.publish(ns).await; + if let Err(e) = super::remove_incomplete_marker(&marker_dir).await { + // The namespace is published and serves; the marker is removed at its next load. It + // matters only if the metastore is lost and rebuilt from the filesystem meanwhile. + tracing::error!(namespace = %name, "could not remove creation marker: {e}"); + } } pub async fn with_authenticated( @@ -322,12 +549,20 @@ impl NamespaceStore { where Fun: FnOnce(&Namespace) -> R, { - if namespace != NamespaceName::default() - && !self.inner.metadata.exists(&namespace).await - && !self.inner.allow_lazy_creation - { - return Err(Error::NamespaceDoesntExist(namespace.to_string())); - } + // `handle` registers the name in the metastore's in-memory map as a side effect, which + // is what `exists` reads: a namespace that must already exist is looked up with the + // side-effect-free `existing_handle`, so that a lookup racing with the end of a failed + // creation cannot bring the name back to life. + let lazily_created = + namespace == NamespaceName::default() || self.inner.allow_lazy_creation; + let handle = if lazily_created { + self.inner.metadata.handle(namespace.clone()).await + } else { + match self.inner.metadata.existing_handle(namespace.clone()).await { + Some(handle) => handle, + None => return Err(Error::NamespaceDoesntExist(namespace.to_string())), + } + }; let f = { let name = namespace.clone(); @@ -341,11 +576,7 @@ impl NamespaceStore { } }; - let handle = self.inner.metadata.handle(namespace.to_owned()).await; - f(self - .load_namespace(&namespace, handle, RestoreOption::Latest) - .await?) - .await + f(self.load_namespace(&namespace, handle).await?).await } fn resolve_attach_fn(&self) -> ResolveNamespacePathFn { @@ -384,31 +615,132 @@ impl NamespaceStore { Ok(ns) } + /// Install `ns` as namespace `name` at the end of an operation that held `entry` (the + /// name's cache entry) write-locked, with `slot` its locked content. + /// + /// Normally `entry` is still the cache's entry for the name and `ns` goes into it. If the + /// cache evicted `entry` meanwhile, `ns` goes into a fresh entry instead and `entry` is + /// left empty: the eviction listener waiting on it then finds nothing to shut down, and + /// lookups that waited on it fall through to the cache (see `load_namespace`). If the entry + /// is gone because the namespace was *destroyed* meanwhile, `ns` goes into `entry` after + /// all: `destroy` is waiting on it and shuts it down. + async fn install( + &self, + name: &NamespaceName, + entry: &NamespaceEntry, + slot: &mut Option, + ns: Namespace, + ) { + match self.inner.store.get(name).await { + Some(current) if Arc::ptr_eq(¤t, entry) => { + slot.replace(ns); + } + _ if !self.inner.metadata.exists(name).await => { + tracing::info!( + namespace = %name, + "namespace was destroyed while an operation on it was in progress" + ); + slot.replace(ns); + } + _ => { + tracing::info!( + namespace = %name, + "namespace entry was evicted while an operation on it was in progress; \ + re-inserting" + ); + self.inner + .store + .insert(name.clone(), Arc::new(RwLock::new(Some(ns)))) + .await; + } + } + } + + /// The entry of an existing namespace, loading it (`RestoreOption::Latest`) if needed. async fn load_namespace( &self, namespace: &NamespaceName, db_config: MetaStoreHandle, - restore_option: RestoreOption, ) -> crate::Result { - let init = async { - let ns = self - .make_namespace(namespace, db_config, restore_option) - .await?; - Ok(Some(ns)) - }; + // An operation in progress on this namespace (create, fork, reset) holds its entry + // locked and pinned: wait on that entry rather than loading the directory it is working + // on (the cache may have evicted the entry meanwhile). If the operation leaves the entry + // empty — it failed, or the entry was evicted and the namespace it published was shut + // down by the eviction listener — decide like any other lookup: the metastore says + // whether the namespace exists, and the cache (re)loads it. + let pinned = self.inner.in_flight.lock().get(namespace).cloned(); + if let Some(entry) = pinned { + tracing::debug!( + namespace = %namespace, + "waiting for an operation in progress on the namespace" + ); + if entry.read().await.is_some() { + return Ok(entry); + } + if !self.inner.metadata.exists(namespace).await { + return Err(Error::NamespaceDoesntExist(namespace.to_string())); + } + } - let before_load = Instant::now(); - let ns = self - .inner - .store - .try_get_with( - namespace.clone(), - init.map_ok(|ns| Arc::new(RwLock::new(ns))), - ) - .await?; - NAMESPACE_LOAD_LATENCY.record(before_load.elapsed()); + let lazily_created = + *namespace == NamespaceName::default() || self.inner.allow_lazy_creation; + // Two rounds at most: an entry that is empty once nobody holds it locked is a leftover + // (of a reservation that was rejected, a reset that failed, a checkpoint of a namespace + // that was not loaded) and would make the namespace unreachable until the cache evicts + // it; drop it and load for real. + for round in 0..2 { + let loaded = AtomicBool::new(false); + let init = async { + loaded.store(true, Ordering::Relaxed); + // The caller checked that the namespace exists; re-check now that we are the + // one loading it, in case it was discarded or destroyed in between (otherwise + // we would bring an empty namespace into existence for a name the metastore + // doesn't know). + if !lazily_created && !self.inner.metadata.exists(namespace).await { + return Err(Error::NamespaceDoesntExist(namespace.to_string())); + } + let ns = self + .make_namespace(namespace, db_config.clone(), RestoreOption::Latest) + .await?; + // A namespace opened from the metastore is complete by definition; this only + // removes a marker left by a crash between persisting the config and removing + // it. + if let Err(e) = ns.mark_complete().await { + tracing::warn!(namespace = %namespace, "could not remove creation marker: {e}"); + } + Ok(Some(ns)) + }; - Ok(ns) + let before_load = Instant::now(); + let entry = self + .inner + .store + .try_get_with( + namespace.clone(), + init.map_ok(|ns| Arc::new(RwLock::new(ns))), + ) + .await?; + if loaded.load(Ordering::Relaxed) { + NAMESPACE_LOAD_LATENCY.record(before_load.elapsed()); + return Ok(entry); + } + + if round == 0 && entry.read().await.is_none() { + let still_cached = self + .inner + .store + .get(namespace) + .await + .is_some_and(|e| Arc::ptr_eq(&e, &entry)); + if still_cached { + tracing::debug!(namespace = %namespace, "dropping empty namespace entry"); + self.inner.store.invalidate(namespace).await; + } + continue; + } + return Ok(entry); + } + unreachable!("the second round always returns") } #[tracing::instrument(skip_all, fields(namespace))] @@ -430,34 +762,88 @@ impl NamespaceStore { .await; }; - // With namespaces disabled, the default namespace can be auto-created, - // otherwise it's an error. + // With namespaces disabled, the default namespace always exists and `create` is an + // idempotent config upsert of it. Creating *from a dump* is not: the dump must never be + // skipped because the namespace happens to be loaded already, so that case takes the + // reserving path below like any other creation. // FIXME: move the default namespace check out of this function. - if self.inner.allow_lazy_creation || namespace == NamespaceName::default() { + if (self.inner.allow_lazy_creation || namespace == NamespaceName::default()) + && matches!(restore_option, RestoreOption::Latest) + { tracing::trace!("auto-creating the namespace"); - } else if self.inner.metadata.exists(&namespace).await { - return Err(Error::NamespaceAlreadyExist(namespace.to_string())); + let handle = self.inner.metadata.handle(namespace.clone()).await; + handle.store(Arc::new(db_config)).await?; + self.load_namespace(&namespace, handle).await?; + return Ok(()); } + // Nothing below is visible or durable until the namespace is complete: + // - the reservation locks the name; dropping it unpublished undoes the in-memory config; + // - the configurator removes a brand-new directory if setup fails or is cancelled; + // - the metastore row is written last, so its presence implies a complete namespace. + let started = Instant::now(); let db_config = Arc::new(db_config); + let reservation = Reservation::acquire(self, namespace.clone(), db_config.clone()).await?; + match self + .create_reserved(&namespace, db_config, restore_option) + .await + { + Ok(ns) => { + self.publish(reservation, ns).await; + tracing::info!( + namespace = %namespace, + elapsed_ms = started.elapsed().as_millis() as u64, + "namespace created" + ); + Ok(()) + } + Err(e) => { + tracing::warn!(namespace = %namespace, error = %e, "namespace creation failed"); + reservation.abort().await; + Err(e) + } + } + } + + /// The part of `create` that runs with the name reserved: everything up to and including + /// the commit point. + async fn create_reserved( + &self, + namespace: &NamespaceName, + db_config: Arc, + restore_option: RestoreOption, + ) -> crate::Result { let handle = self.inner.metadata.handle(namespace.clone()).await; - tracing::debug!("storing db config"); - handle.store(db_config).await?; - tracing::debug!("completed storing db config, loading namespace"); - self.load_namespace(&namespace, handle, restore_option) + // In memory only: setup reads it; nothing is persisted until setup succeeded. + handle.store_and_maybe_flush(Some(db_config), false).await?; + // whatever a creation of this name that never completed left behind + self.get_configurator(&handle.get()) + .discard_incomplete(namespace)?; + tracing::debug!("setting up namespace"); + let ns = self + .make_namespace(namespace, handle.clone(), restore_option) .await?; - - tracing::debug!("completed loading namespace"); - - Ok(()) + if self.inner.has_shutdown.load(Ordering::Relaxed) { + return Err(Error::NamespaceStoreShutdown); + } + tracing::debug!("storing db config"); + handle.flush().await?; // commit point + Ok(ns) } pub async fn shutdown(self) -> crate::Result<()> { let mut set = JoinSet::new(); self.inner.has_shutdown.store(true, Ordering::Relaxed); - for (_name, entry) in self.inner.store.iter() { + for (name, entry) in self.inner.store.iter() { let snapshow_at_shutdown = self.inner.snapshot_at_shutdown; + if self.inner.in_flight.lock().contains_key(&name) { + // A creation, fork or reset holds this entry locked, possibly for minutes. It + // notices the shutdown itself (`has_shutdown`) and discards its work; waiting + // here would hold up the shutdown of every other namespace. + tracing::warn!(namespace = %name, "operation in progress at shutdown; not waiting for it"); + continue; + } let mut lock = entry.write().await; if let Some(ns) = lock.take() { set.spawn(async move { @@ -532,3 +918,419 @@ impl NamespaceStore { .await } } + +#[cfg(test)] +mod test { + use std::path::Path; + + use bytes::Bytes; + use tokio::sync::{mpsc, Semaphore}; + use tokio_stream::wrappers::ReceiverStream; + + use crate::connection::Connection as _; + use crate::namespace::configurator::{ + BaseNamespaceConfig, NamespaceConfigurators, PrimaryConfig, PrimaryConfigurator, + }; + use crate::namespace::dump_import::DumpImporterKind; + use crate::namespace::meta_store::{metastore_connection_maker, MetaStore}; + use crate::namespace::DumpSource; + + use super::*; + + async fn make_store(path: &Path) -> NamespaceStore { + let (maker, manager) = metastore_connection_maker(None, path).await.unwrap(); + let meta_store = MetaStore::new( + Default::default(), + path, + maker().unwrap(), + manager, + DatabaseKind::Primary, + ) + .await + .unwrap(); + let base_config = BaseNamespaceConfig { + base_path: path.to_path_buf().into(), + extensions: Arc::new([]), + stats_sender: mpsc::channel(1).0, + max_response_size: 100000000000000, + max_total_response_size: 100000000000, + max_concurrent_connections: Arc::new(Semaphore::new(10)), + max_concurrent_requests: 10000, + encryption_config: None, + connection_creation_timeout: None, + disable_intelligent_throttling: false, + dump_import: Default::default(), + }; + let primary_config = PrimaryConfig { + max_log_size: 1000000000, + max_log_duration: None, + bottomless_replication: None, + scripted_backup: None, + checkpoint_interval: None, + }; + let mut configurators = NamespaceConfigurators::empty(); + configurators.with_primary(PrimaryConfigurator::new( + base_config, + primary_config, + Arc::new(libsql_sys::wal::Sqlite3WalManager::default), + )); + NamespaceStore::new( + false, + false, + 10, + meta_store, + configurators, + DatabaseKind::Primary, + ) + .await + .unwrap() + } + + /// The reservation is taken a few awaits into `create`; wait for it rather than guess. + async fn wait_until_in_flight(store: &NamespaceStore, name: &NamespaceName) { + for _ in 0..500 { + if store.inner.in_flight.lock().contains_key(name) { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("creation of `{name}` never reserved the name"); + } + + fn slow_dump() -> (mpsc::Sender>, DumpSource) { + let (tx, rx) = mpsc::channel::>(8); + let dump = DumpSource::new(Box::new(ReceiverStream::new(rx))) + .with_importer(Some(DumpImporterKind::Streaming)); + (tx, dump) + } + + async fn count_test_rows(store: &NamespaceStore, ns: &str) -> crate::Result { + let maker = store + .with(NamespaceName::from_string(ns.into()).unwrap(), |ns| { + ns.db.connection_maker() + }) + .await?; + let conn = maker.create().await?; + Ok(conn.with_raw(|c| c.query_row("select count(*) from test", [], |r| r.get(0)))?) + } + + /// The cache evicts the entry of a namespace while it is being created (here: explicitly + /// invalidated, which takes the same path). A lookup issued afterwards must still wait for + /// the creation and then see the imported data, instead of lazily opening the directory the + /// import is writing to. + #[tokio::test] + async fn lookup_waits_for_creation_even_if_the_entry_was_evicted() { + let tmp = tempfile::tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let foo = NamespaceName::from_string("foo".into()).unwrap(); + + let (tx, rx) = mpsc::channel::>(8); + let dump = DumpSource::new(Box::new(ReceiverStream::new(rx))) + .with_importer(Some(DumpImporterKind::Streaming)); + let create = tokio::spawn({ + let store = store.clone(); + let foo = foo.clone(); + async move { + store + .create(foo, RestoreOption::Dump(dump), Default::default()) + .await + } + }); + tx.send(Ok(Bytes::from_static( + b"BEGIN TRANSACTION; CREATE TABLE test (x); INSERT INTO test VALUES (1);", + ))) + .await + .unwrap(); + wait_until_in_flight(&store, &foo).await; + assert!(!create.is_finished()); + + // "eviction" of the reserved entry + store.inner.store.invalidate(&foo).await; + store.inner.store.run_pending_tasks().await; + assert!(!store.inner.store.contains_key(&foo)); + + let lookup = tokio::spawn({ + let store = store.clone(); + async move { count_test_rows(&store, "foo").await } + }); + tokio::time::sleep(Duration::from_millis(200)).await; + assert!( + !lookup.is_finished(), + "the lookup must wait for the creation, not load the directory being imported into" + ); + + tx.send(Ok(Bytes::from_static(b"COMMIT;"))).await.unwrap(); + drop(tx); + create.await.unwrap().unwrap(); + assert_eq!(lookup.await.unwrap().unwrap(), 1); + // the namespace was published into a fresh cache entry and is served normally + assert!(store.inner.store.contains_key(&foo)); + assert_eq!(count_test_rows(&store, "foo").await.unwrap(), 1); + assert!(store.inner.in_flight.lock().is_empty()); + } + + /// Same situation, but the creation fails: the waiting lookup must report that the + /// namespace does not exist, and nothing must be left behind. + #[tokio::test] + async fn lookup_sees_failure_of_creation_even_if_the_entry_was_evicted() { + let tmp = tempfile::tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let foo = NamespaceName::from_string("foo".into()).unwrap(); + + let (tx, rx) = mpsc::channel::>(8); + let dump = DumpSource::new(Box::new(ReceiverStream::new(rx))) + .with_importer(Some(DumpImporterKind::Streaming)); + let create = tokio::spawn({ + let store = store.clone(); + let foo = foo.clone(); + async move { + store + .create(foo, RestoreOption::Dump(dump), Default::default()) + .await + } + }); + tx.send(Ok(Bytes::from_static( + b"BEGIN TRANSACTION; CREATE TABLE test (x); INSERT INTO test VALUES (1);", + ))) + .await + .unwrap(); + wait_until_in_flight(&store, &foo).await; + + store.inner.store.invalidate(&foo).await; + store.inner.store.run_pending_tasks().await; + + let lookup = tokio::spawn({ + let store = store.clone(); + async move { count_test_rows(&store, "foo").await } + }); + tokio::time::sleep(Duration::from_millis(200)).await; + assert!(!lookup.is_finished()); + + tx.send(Ok(Bytes::from_static(b"THIS IS NOT SQL;"))) + .await + .unwrap(); + drop(tx); + assert!(create.await.unwrap().is_err()); + assert!(matches!( + lookup.await.unwrap(), + Err(Error::NamespaceDoesntExist(_)) + )); + assert!(!store.exists(&foo).await); + // the cleanup runs on the blocking pool; give it a moment + for _ in 0..50 { + if !tmp.path().join("dbs").join("foo").exists() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(!tmp.path().join("dbs").join("foo").exists()); + assert!(store.inner.in_flight.lock().is_empty()); + } + + /// `destroy` is issued while the namespace is still being created. It must wait for the + /// creation and destroy its result: nothing left in the cache, on disk or in the metastore, + /// and the name can be created again. (Before: the result could be re-inserted into the + /// cache as a live namespace whose directory `destroy` had just removed.) + #[tokio::test] + async fn destroy_during_creation_destroys_the_result() { + let tmp = tempfile::tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let foo = NamespaceName::from_string("foo".into()).unwrap(); + + let (tx, dump) = slow_dump(); + let create = tokio::spawn({ + let (store, foo) = (store.clone(), foo.clone()); + async move { + store + .create(foo, RestoreOption::Dump(dump), Default::default()) + .await + } + }); + tx.send(Ok(Bytes::from_static( + b"BEGIN TRANSACTION; CREATE TABLE test (x); INSERT INTO test VALUES (1);", + ))) + .await + .unwrap(); + wait_until_in_flight(&store, &foo).await; + // make it the interesting case: the cache no longer holds the reserved entry + store.inner.store.invalidate(&foo).await; + store.inner.store.run_pending_tasks().await; + + let destroy = tokio::spawn({ + let (store, foo) = (store.clone(), foo.clone()); + async move { store.destroy(foo, true).await } + }); + tokio::time::sleep(Duration::from_millis(100)).await; + assert!(!destroy.is_finished(), "destroy must wait for the creation"); + + tx.send(Ok(Bytes::from_static(b"COMMIT;"))).await.unwrap(); + drop(tx); + create.await.unwrap().unwrap(); + destroy.await.unwrap().unwrap(); + + assert!(!store.exists(&foo).await); + assert!(!store.inner.store.contains_key(&foo)); + assert!(!tmp.path().join("dbs").join("foo").exists()); + assert!(store.inner.in_flight.lock().is_empty()); + assert!(matches!( + count_test_rows(&store, "foo").await, + Err(Error::NamespaceDoesntExist(_)) + )); + + // and the name is free again + let (tx, dump) = slow_dump(); + let create = tokio::spawn({ + let (store, foo) = (store.clone(), foo.clone()); + async move { + store + .create(foo, RestoreOption::Dump(dump), Default::default()) + .await + } + }); + tx.send(Ok(Bytes::from_static( + b"BEGIN TRANSACTION; CREATE TABLE test (x); INSERT INTO test VALUES (1); COMMIT;", + ))) + .await + .unwrap(); + drop(tx); + create.await.unwrap().unwrap(); + assert_eq!(count_test_rows(&store, "foo").await.unwrap(), 1); + } + + /// A failed creation is undone before the name is released, so a retry issued the moment + /// the failure is reported neither races the cleanup nor finds leftovers. + #[tokio::test] + async fn retry_right_after_failure_starts_clean() { + let tmp = tempfile::tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let foo = NamespaceName::from_string("foo".into()).unwrap(); + let dir = tmp.path().join("dbs").join("foo"); + + for attempt in 0..3 { + let (tx, dump) = slow_dump(); + tx.send(Ok(Bytes::from_static( + b"BEGIN TRANSACTION; CREATE TABLE test (x); INSERT INTO test VALUES (1); NOT SQL;", + ))) + .await + .unwrap(); + drop(tx); + let res = store + .create(foo.clone(), RestoreOption::Dump(dump), Default::default()) + .await; + assert!(res.is_err(), "attempt {attempt}"); + // no waiting: the path is free and the name unknown as soon as `create` returns + assert!(!dir.exists(), "attempt {attempt}"); + assert!(!store.exists(&foo).await, "attempt {attempt}"); + } + + let (tx, dump) = slow_dump(); + tx.send(Ok(Bytes::from_static( + b"BEGIN TRANSACTION; CREATE TABLE test (x); INSERT INTO test VALUES (1); COMMIT;", + ))) + .await + .unwrap(); + drop(tx); + store + .create(foo.clone(), RestoreOption::Dump(dump), Default::default()) + .await + .unwrap(); + assert_eq!(count_test_rows(&store, "foo").await.unwrap(), 1); + assert!(!dir + .join(crate::namespace::incomplete::INCOMPLETE_MARKER) + .exists()); + } + + /// The config of a namespace linked to a shared schema cannot be *removed* while the schema + /// has pending migration jobs — but it must always be possible to *discard* it when its + /// creation did not complete, or the name would be stuck (`exists` true, no row, no data). + #[tokio::test] + async fn discard_ignores_the_guards_of_remove() { + let tmp = tempfile::tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let linked = NamespaceName::from_string("linked".into()).unwrap(); + let handle = store.inner.metadata.handle(linked.clone()).await; + handle + .store_and_maybe_flush( + Some(Arc::new(DatabaseConfig { + shared_schema_name: Some(NamespaceName::from_string("schema".into()).unwrap()), + ..Default::default() + })), + false, + ) + .await + .unwrap(); + assert!(store.exists(&linked).await); + + let metadata = store.inner.metadata.clone(); + let name = linked.clone(); + tokio::task::spawn_blocking(move || metadata.discard_blocking(name)) + .await + .unwrap() + .unwrap(); + assert!(!store.exists(&linked).await); + } + + /// The creating task is aborted mid-import (the request was cancelled). The reservation is + /// dropped unpublished: config discarded, directory discarded, name released — and a + /// lookup that was waiting sees "doesn't exist". + #[tokio::test] + async fn aborted_creation_is_undone() { + let tmp = tempfile::tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let foo = NamespaceName::from_string("foo".into()).unwrap(); + let dir = tmp.path().join("dbs").join("foo"); + + let (tx, dump) = slow_dump(); + let create = tokio::spawn({ + let (store, foo) = (store.clone(), foo.clone()); + async move { + store + .create(foo, RestoreOption::Dump(dump), Default::default()) + .await + } + }); + tx.send(Ok(Bytes::from_static( + b"BEGIN TRANSACTION; CREATE TABLE test (x); INSERT INTO test VALUES (1);", + ))) + .await + .unwrap(); + wait_until_in_flight(&store, &foo).await; + assert!(dir + .join(crate::namespace::incomplete::INCOMPLETE_MARKER) + .exists()); + + let lookup = tokio::spawn({ + let store = store.clone(); + async move { count_test_rows(&store, "foo").await } + }); + tokio::time::sleep(Duration::from_millis(100)).await; + assert!(!lookup.is_finished()); + + create.abort(); + assert!(create.await.unwrap_err().is_cancelled()); + + assert!(matches!( + lookup.await.unwrap(), + Err(Error::NamespaceDoesntExist(_)) + )); + // the cleanup ran on the blocking pool; the lookup's completion proves the lock was + // released after it, so by now everything is undone + assert!(!store.exists(&foo).await); + assert!(!dir.exists()); + assert!(store.inner.in_flight.lock().is_empty()); + + // the importer's executor thread is still unwinding its channel; the name is free anyway + let (tx, dump) = slow_dump(); + tx.send(Ok(Bytes::from_static( + b"BEGIN TRANSACTION; CREATE TABLE test (x); INSERT INTO test VALUES (1); COMMIT;", + ))) + .await + .unwrap(); + drop(tx); + store + .create(foo.clone(), RestoreOption::Dump(dump), Default::default()) + .await + .unwrap(); + assert_eq!(count_test_rows(&store, "foo").await.unwrap(), 1); + } +} diff --git a/libsql-server/tests/namespaces/dumps.rs b/libsql-server/tests/namespaces/dumps.rs index 958d24daab..c9410c392c 100644 --- a/libsql-server/tests/namespaces/dumps.rs +++ b/libsql-server/tests/namespaces/dumps.rs @@ -17,11 +17,11 @@ use crate::common::http::{Client, Response}; use crate::common::net::{TurmoilAcceptor, TurmoilConnector}; use crate::namespaces::{make_primary, make_primary_with_db_config}; -const BUFFERED: Option<&str> = Some("buffered"); -const STREAMING: Option<&str> = Some("streaming"); +pub(super) const BUFFERED: Option<&str> = Some("buffered"); +pub(super) const STREAMING: Option<&str> = Some("streaming"); /// `POST /v1/namespaces/:ns/create` with `dump_url` and, optionally, `dump_importer`. -async fn create_from_dump( +pub(super) async fn create_from_dump( client: &Client, ns: &str, dump_url: &str, @@ -39,17 +39,22 @@ async fn create_from_dump( .await } -fn file_url(path: &Path) -> String { +pub(super) fn file_url(path: &Path) -> String { format!("file:{}", path.display()) } -fn sim() -> Sim<'static> { +pub(super) fn sim() -> Sim<'static> { Builder::new() .simulation_duration(Duration::from_secs(1000)) + // Dumps are served in many small chunks that the server only starts reading once the + // namespace is set up (fsyncs, file creation — real time, during which turmoil keeps + // advancing simulated time and the dump store keeps writing). Real TCP would apply + // back-pressure; turmoil panics once its socket buffer (64 segments by default) is full. + .tcp_capacity(4096) .build() } -async fn count_rows(ns: &str, table: &str) -> anyhow::Result { +pub(super) async fn count_rows(ns: &str, table: &str) -> anyhow::Result { let db = Database::open_remote_with_connector( &format!("http://{ns}.primary:8080"), "", @@ -69,7 +74,7 @@ fn make_dump_store(sim: &mut Sim, chunks: Vec>) { } /// Like [`make_dump_store`], pausing `pause` (simulated time) before each chunk. -fn make_dump_store_paced( +pub(super) fn make_dump_store_paced( sim: &mut Sim, chunks: Vec>, pause: Duration, @@ -124,7 +129,39 @@ fn make_dump_store_chunked(sim: &mut Sim, dump: &'static str, chunk_size: usize) ); } -const SIMPLE_DUMP: &str = r#" +/// Serve `dump` from `dump-store:8080` in 64-byte chunks, 100ms (simulated) apart: with +/// turmoil's up-to-100ms message latency this keeps a creation in flight for a couple of +/// seconds, long enough for other requests to be issued while it runs. +pub(super) fn make_slow_dump_store(sim: &mut Sim, dump: &'static str) { + make_dump_store_paced( + sim, + dump.as_bytes() + .chunks(64) + .map(|c| Ok(Bytes::copy_from_slice(c))) + .collect(), + Duration::from_millis(100), + ); +} + +/// Ten rows, importable by both importers (no "attach" anywhere), ~1.1 KB so that +/// [`make_slow_dump_store`] delivers it in ~18 chunks. +pub(super) const SLOW_DUMP: &str = r#"PRAGMA foreign_keys=OFF; +BEGIN TRANSACTION; +CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT, note TEXT); +INSERT INTO test VALUES (1, 'one', 'the first row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (2, 'two', 'the second row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (3, 'three', 'the third row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (4, 'four', 'the fourth row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (5, 'five', 'the fifth row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (6, 'six', 'the sixth row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (7, 'seven', 'the seventh row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (8, 'eight', 'the eighth row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (9, 'nine', 'the ninth row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (10, 'ten', 'the tenth row of a dump that takes a while to arrive'); +COMMIT; +"#; + +pub(super) const SIMPLE_DUMP: &str = r#" PRAGMA foreign_keys=OFF; BEGIN TRANSACTION; CREATE TABLE test (x); @@ -745,7 +782,7 @@ fn load_dump_with_nested_case_streaming() { /// Exercises framing across arbitrary HTTP body chunk boundaries, including inside multibyte /// characters, string literals with semicolons, comments and a trigger body. -const CHUNKY_DUMP: &str = r#"PRAGMA foreign_keys=OFF; +pub(super) const CHUNKY_DUMP: &str = r#"PRAGMA foreign_keys=OFF; BEGIN TRANSACTION; -- a comment; with a semicolon CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT); @@ -1437,68 +1474,88 @@ fn streaming_failure_under_backpressure() { sim.run().unwrap(); } -/// The admin request is abandoned while the import is still streaming in. The executor thread -/// must notice the closed channel, roll back and release the connection, so the server stays -/// healthy and the namespace can be used afterwards instead of being wedged by a half-open -/// transaction. -#[test] -fn streaming_cancelled_request_rolls_back() { +/// The admin request is abandoned while the dump is still being transferred. The namespace +/// must not survive in any form: the importer rolls back and releases its connection, the +/// configurator removes the directory and the store never persists the config, so the name is +/// immediately reusable (see `docs/ATOMIC_NAMESPACE_CREATE_DESIGN.md`). +fn cancelled_create_leaves_no_trace_with(importer: Option<&'static str>) { let mut sim = sim(); let tmp = tempdir().unwrap(); + let tmp_path = tmp.path().to_path_buf(); make_primary(&mut sim, tmp.path().to_path_buf()); - // Deliver the dump slowly (64-byte chunks, 50ms apart) so the request is still in flight - // when it is abandoned. Few, large chunks also keep the number of unread segments below - // turmoil's simulated socket buffer once nobody reads them anymore. - make_dump_store_paced( - &mut sim, - CHUNKY_DUMP - .as_bytes() - .chunks(64) - .map(|c| Ok(Bytes::copy_from_slice(c))) - .collect(), - Duration::from_millis(50), - ); + // Deliver the dump slowly so the request is still in flight when it is abandoned. Few, + // large chunks also keep the number of unread segments below turmoil's simulated socket + // buffer once nobody reads them anymore. + make_slow_dump_store(&mut sim, SLOW_DUMP); sim.client("client", async move { let client = Client::new(); + let dir = tmp_path.join("dbs").join("foo"); + // Cancel once the creation is provably under way: the namespace directory exists and + // is marked incomplete (turmoil's random message latency makes a fixed delay either + // too early or too late). + let marker = dir.join(super::lifecycle::INCOMPLETE_MARKER); + let request = create_from_dump(&client, "foo", "http://dump-store:8080/", importer); let aborted = tokio::time::timeout( - Duration::from_millis(100), - create_from_dump(&client, "foo", "http://dump-store:8080/", STREAMING), + Duration::from_secs(10), + futures::future::select( + Box::pin(request), + Box::pin(async { + while !marker.exists() { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }), + ), ) - .await; - assert!(aborted.is_err(), "the request should still be in flight"); + .await + .expect("the creation should have started"); + assert!( + matches!(aborted, futures::future::Either::Right(_)), + "the request should still be in flight" + ); + // `aborted` holds the request future: dropping it is the cancellation. + drop(aborted); // Let the server notice the closed connection (simulated time) and the executor thread // roll back and drop its connection (real time: the sim clock doesn't wait for it). tokio::time::sleep(Duration::from_secs(2)).await; std::thread::sleep(Duration::from_millis(300)); - // Depending on where the request was cut, the namespace was registered (pre-existing - // behavior: the metastore row survives a failed create) or not. Either way it must be - // usable now: no lingering write transaction, no partial data. + // The namespace does not exist, for the admin API, for users, and on disk. let resp = client - .post_raw("http://primary:9090/v1/namespaces/foo/create", json!({})) + .get("http://primary:9090/v1/namespaces/foo/config") .await?; - assert!( - resp.status() == StatusCode::OK || resp.status() == StatusCode::BAD_REQUEST, - "unexpected status {}", - resp.status() - ); - assert!(count_rows("foo", "test").await.is_err()); + assert_eq!(resp.status(), StatusCode::NOT_FOUND); + let err = count_rows("foo", "test").await.unwrap_err().to_string(); + assert!(err.contains("doesn't exist"), "unexpected error: {err}"); + super::lifecycle::wait_until_gone(&dir); + + // ... so the same name can be created again, from the same dump, and used. + let resp = create_from_dump(&client, "foo", "http://dump-store:8080/", importer).await?; + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(count_rows("foo", "test").await?, 10); let db = Database::open_remote_with_connector("http://foo.primary:8080", "", TurmoilConnector)?; let conn = db.connect()?; - conn.execute("create table after_cancel (x)", ()).await?; - conn.execute("insert into after_cancel values (1)", ()) - .await?; - assert_eq!(count_rows("foo", "after_cancel").await?, 1); - - // and other namespaces are unaffected - let resp = create_from_dump(&client, "bar", "http://dump-store:8080/", STREAMING).await?; - assert_eq!(resp.status(), StatusCode::OK); - assert_eq!(count_rows("bar", "test").await?, 3); + conn.execute( + "insert into test values (11, 'eleven', 'written after the cancelled attempt')", + (), + ) + .await?; + assert_eq!(count_rows("foo", "test").await?, 11); + assert!(!marker.exists()); Ok(()) }); sim.run().unwrap(); } + +#[test] +fn cancelled_create_leaves_no_trace() { + cancelled_create_leaves_no_trace_with(BUFFERED); +} + +#[test] +fn cancelled_create_leaves_no_trace_streaming() { + cancelled_create_leaves_no_trace_with(STREAMING); +} diff --git a/libsql-server/tests/namespaces/lifecycle.rs b/libsql-server/tests/namespaces/lifecycle.rs new file mode 100644 index 0000000000..37e53fbcb4 --- /dev/null +++ b/libsql-server/tests/namespaces/lifecycle.rs @@ -0,0 +1,389 @@ +//! Namespace creation is all-or-nothing: after a failed, cancelled or crashed +//! `POST /v1/namespaces/:ns/create` the namespace either does not exist (no config, no +//! directory, name reusable) or is complete. See `docs/ATOMIC_NAMESPACE_CREATE_DESIGN.md`. + +use std::path::Path; +use std::time::Duration; + +use hyper::StatusCode; +use serde_json::json; +use tempfile::tempdir; + +use crate::common::http::Client; +use crate::namespaces::dumps::{ + count_rows, create_from_dump, file_url, make_slow_dump_store, sim, BUFFERED, SIMPLE_DUMP, + SLOW_DUMP, STREAMING, +}; +use crate::namespaces::{make_primary, make_single_namespace_primary}; + +pub(super) const INCOMPLETE_MARKER: &str = ".incomplete"; + +const BROKEN_DUMP: &str = r#" + BEGIN TRANSACTION; + CREATE TABLE test (x); + INSERT INTO test VALUES(1); + THIS IS NOT SQL; + COMMIT;"#; + +/// [`SLOW_DUMP`] with a syntax error right before the end. +const SLOW_BROKEN_DUMP: &str = r#"PRAGMA foreign_keys=OFF; +BEGIN TRANSACTION; +CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT, note TEXT); +INSERT INTO test VALUES (1, 'one', 'the first row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (2, 'two', 'the second row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (3, 'three', 'the third row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (4, 'four', 'the fourth row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (5, 'five', 'the fifth row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (6, 'six', 'the sixth row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (7, 'seven', 'the seventh row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (8, 'eight', 'the eighth row of a dump that takes a while to arrive'); +INSERT INTO test VALUES (9, 'nine', 'the ninth row of a dump that takes a while to arrive'); +THIS IS NOT SQL; +COMMIT; +"#; + +/// Directory removal after an aborted creation is handed to the runtime; poll (real time, the +/// blocking pool is not driven by the simulated clock) until it is gone. +pub(super) fn wait_until_gone(path: &Path) { + for _ in 0..100 { + if !path.exists() { + return; + } + std::thread::sleep(Duration::from_millis(20)); + } + panic!("{} still exists", path.display()); +} + +/// After a creation *failed* (as opposed to being cancelled) nothing of it is left by the time +/// the error is reported: no polling here. +async fn assert_absent(client: &Client, tmp: &Path, ns: &str) -> anyhow::Result<()> { + assert!( + !tmp.join("dbs").join(ns).exists(), + "{ns} directory left behind" + ); + let resp = client + .get(&format!("http://primary:9090/v1/namespaces/{ns}/config")) + .await?; + assert_eq!( + resp.status(), + StatusCode::NOT_FOUND, + "{ns} should not exist" + ); + let err = count_rows(ns, "test").await.unwrap_err().to_string(); + assert!(err.contains("doesn't exist"), "unexpected error: {err}"); + Ok(()) +} + +/// A dump that fails to import leaves nothing behind, and the name can be reused right away. +/// Before this change the config survived and the next access lazily created an empty database. +fn failed_create_leaves_no_trace_with(importer: Option<&'static str>, shared_schema: bool) { + let mut sim = sim(); + let tmp = tempdir().unwrap(); + let tmp_path = tmp.path().to_path_buf(); + std::fs::write(tmp_path.join("broken.sql"), BROKEN_DUMP).unwrap(); + std::fs::write(tmp_path.join("good.sql"), SIMPLE_DUMP).unwrap(); + make_primary(&mut sim, tmp.path().to_path_buf()); + + sim.client("client", async move { + let client = Client::new(); + let create = |dump: &str| { + let mut body = json!({ "dump_url": file_url(&tmp_path.join(dump)), "shared_schema": shared_schema }); + if let Some(importer) = importer { + body["dump_importer"] = json!(importer); + } + client.post_raw("http://primary:9090/v1/namespaces/foo/create", body) + }; + + let resp = create("broken.sql").await?; + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + assert_absent(&client, &tmp_path, "foo").await?; + + let resp = create("good.sql").await?; + assert_eq!( + resp.status(), + StatusCode::OK, + "{}", + resp.body_string().await.unwrap_or_default() + ); + assert_eq!(count_rows("foo", "test").await?, 1); + assert!(!tmp_path.join("dbs/foo").join(INCOMPLETE_MARKER).exists()); + Ok(()) + }); + + sim.run().unwrap(); +} + +#[test] +fn failed_create_leaves_no_trace() { + failed_create_leaves_no_trace_with(BUFFERED, false); +} + +#[test] +fn failed_create_leaves_no_trace_streaming() { + failed_create_leaves_no_trace_with(STREAMING, false); +} + +#[test] +fn failed_create_leaves_no_trace_shared_schema() { + failed_create_leaves_no_trace_with(STREAMING, true); +} + +/// Requests for a namespace that is being created wait for the outcome: a second `create` +/// for the same name is rejected once the first one succeeded, and a user request issued +/// mid-import sees the complete data. +#[test] +fn concurrent_requests_wait_for_creation() { + let mut sim = sim(); + let tmp = tempdir().unwrap(); + make_primary(&mut sim, tmp.path().to_path_buf()); + make_slow_dump_store(&mut sim, SLOW_DUMP); + + sim.client("client", async move { + let client = Client::new(); + let slow = create_from_dump(&client, "foo", "http://dump-store:8080/", STREAMING); + // issued while the import is in progress (it takes ~2s of simulated time) + let second = async { + tokio::time::sleep(Duration::from_millis(500)).await; + client + .post_raw("http://primary:9090/v1/namespaces/foo/create", json!({})) + .await + }; + let user = async { + tokio::time::sleep(Duration::from_millis(500)).await; + count_rows("foo", "test").await + }; + let (slow, second, user) = tokio::join!(slow, second, user); + assert_eq!(slow?.status(), StatusCode::OK); + let second = second?; + assert_eq!(second.status(), StatusCode::BAD_REQUEST); + assert!(second.body_string().await?.contains("already exists")); + assert_eq!(user?, 10); + Ok(()) + }); + + sim.run().unwrap(); +} + +/// If the creation that others are waiting on fails, the waiters observe a namespace that does +/// not exist, and a subsequent `create` of the same name succeeds. +#[test] +fn concurrent_create_succeeds_after_failure() { + let mut sim = sim(); + let tmp = tempdir().unwrap(); + let tmp_path = tmp.path().to_path_buf(); + std::fs::write(tmp_path.join("good.sql"), SIMPLE_DUMP).unwrap(); + make_primary(&mut sim, tmp.path().to_path_buf()); + make_slow_dump_store(&mut sim, SLOW_BROKEN_DUMP); + + sim.client("client", async move { + let client = Client::new(); + let slow = create_from_dump(&client, "foo", "http://dump-store:8080/", STREAMING); + // issued while the import is in progress (it takes ~2s of simulated time) + let user = async { + tokio::time::sleep(Duration::from_millis(500)).await; + count_rows("foo", "test").await + }; + let (slow, user) = tokio::join!(slow, user); + assert_eq!(slow?.status(), StatusCode::BAD_REQUEST); + let err = user.unwrap_err().to_string(); + assert!(err.contains("doesn't exist"), "unexpected error: {err}"); + assert_absent(&client, &tmp_path, "foo").await?; + + let resp = create_from_dump( + &client, + "foo", + &file_url(&tmp_path.join("good.sql")), + STREAMING, + ) + .await?; + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(count_rows("foo", "test").await?, 1); + Ok(()) + }); + + sim.run().unwrap(); +} + +/// A directory left by a creation the process died in the middle of (it still carries the +/// marker, there is no config for it) is discarded by the next creation of that name, and is +/// not adopted as a namespace when the metastore is rebuilt from the filesystem. +#[test] +fn crash_remnant_is_discarded() { + let mut sim = sim(); + let tmp = tempdir().unwrap(); + let tmp_path = tmp.path().to_path_buf(); + std::fs::write(tmp_path.join("good.sql"), SIMPLE_DUMP).unwrap(); + let remnant = tmp_path.join("dbs/foo"); + std::fs::create_dir_all(&remnant).unwrap(); + for f in [INCOMPLETE_MARKER, ".sentinel", "data", "wallog"] { + std::fs::write(remnant.join(f), b"junk").unwrap(); + } + make_primary(&mut sim, tmp.path().to_path_buf()); + + sim.client("client", async move { + let client = Client::new(); + // not recovered as a namespace at startup (the metastore was empty) + let resp = client + .get("http://primary:9090/v1/namespaces/foo/config") + .await?; + assert_eq!(resp.status(), StatusCode::NOT_FOUND); + + let resp = create_from_dump( + &client, + "foo", + &file_url(&tmp_path.join("good.sql")), + STREAMING, + ) + .await?; + assert_eq!( + resp.status(), + StatusCode::OK, + "{}", + resp.body_string().await.unwrap_or_default() + ); + assert_eq!(count_rows("foo", "test").await?, 1); + assert!(!remnant.join(INCOMPLETE_MARKER).exists()); + Ok(()) + }); + + sim.run().unwrap(); +} + +/// Directories without the marker are never touched by a creation, whatever is in them. +#[test] +fn unmarked_directory_is_left_alone() { + let mut sim = sim(); + let tmp = tempdir().unwrap(); + let tmp_path = tmp.path().to_path_buf(); + std::fs::write(tmp_path.join("good.sql"), SIMPLE_DUMP).unwrap(); + let dir = tmp_path.join("dbs/foo"); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write(dir.join("keep.txt"), b"precious").unwrap(); + make_primary(&mut sim, tmp.path().to_path_buf()); + + sim.client("client", async move { + let client = Client::new(); + // No database in there, so the import proceeds into the existing directory ... + let resp = create_from_dump( + &client, + "foo", + &file_url(&tmp_path.join("good.sql")), + STREAMING, + ) + .await?; + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(count_rows("foo", "test").await?, 1); + // ... and whatever was in it is left alone, before and after. + assert_eq!(std::fs::read(dir.join("keep.txt"))?, b"precious"); + assert!(!dir.join(INCOMPLETE_MARKER).exists()); + Ok(()) + }); + + sim.run().unwrap(); +} + +/// A marker on a namespace that *is* known to the metastore (the process died between +/// persisting the config and removing the marker) is harmless: `create` rejects the name, +/// the data is served, and the marker is dropped when the namespace is next loaded. +#[test] +fn marker_on_published_namespace_is_ignored() { + let tmp = tempdir().unwrap(); + let tmp_path = tmp.path().to_path_buf(); + std::fs::write(tmp_path.join("good.sql"), SIMPLE_DUMP).unwrap(); + let marker = tmp_path.join("dbs/foo").join(INCOMPLETE_MARKER); + + let mut first = sim(); + make_primary(&mut first, tmp.path().to_path_buf()); + first.client("client", { + let tmp_path = tmp_path.clone(); + let marker = marker.clone(); + async move { + let client = Client::new(); + let resp = create_from_dump( + &client, + "foo", + &file_url(&tmp_path.join("good.sql")), + STREAMING, + ) + .await?; + assert_eq!(resp.status(), StatusCode::OK); + assert!(!marker.exists()); + + // simulate a crash right after the config was persisted + std::fs::write(&marker, b"").unwrap(); + let resp = create_from_dump( + &client, + "foo", + &file_url(&tmp_path.join("good.sql")), + STREAMING, + ) + .await?; + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + assert!(resp.body_string().await?.contains("already exists")); + assert_eq!(count_rows("foo", "test").await?, 1); + assert!(marker.exists(), "the marker is only removed on load"); + Ok(()) + } + }); + first.run().unwrap(); + drop(first); // the first server must be gone before the second opens the same files + + // restart: nothing is loaded yet, the marker is still there and the metastore knows the + // name. A `create` must be rejected on the strength of the metastore alone (not wipe the + // directory); then the namespace loads and the stale marker goes away. + let mut restarted = sim(); + make_primary(&mut restarted, tmp.path().to_path_buf()); + restarted.client("client", async move { + let client = Client::new(); + let resp = create_from_dump( + &client, + "foo", + &file_url(&tmp_path.join("good.sql")), + STREAMING, + ) + .await?; + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + assert!(marker.exists(), "rejected before touching the directory"); + assert_eq!(count_rows("foo", "test").await?, 1); + wait_until_gone(&marker); + Ok(()) + }); + restarted.run().unwrap(); +} + +/// Single-namespace mode: `create` of the default namespace is a config upsert, but creating +/// it *from a dump* while it already exists must be rejected rather than silently skipping the +/// import (which is what happened before). +#[test] +fn single_namespace_mode_rejects_dump_into_existing_namespace() { + let mut sim = sim(); + let tmp = tempdir().unwrap(); + let tmp_path = tmp.path().to_path_buf(); + std::fs::write(tmp_path.join("good.sql"), SIMPLE_DUMP).unwrap(); + make_single_namespace_primary(&mut sim, tmp.path().to_path_buf()); + + sim.client("client", async move { + let client = Client::new(); + // config upsert of the (already created) default namespace still works + let resp = client + .post_raw( + "http://primary:9090/v1/namespaces/default/create", + json!({ "max_db_size": "1mb" }), + ) + .await?; + assert_eq!(resp.status(), StatusCode::OK); + + let resp = create_from_dump( + &client, + "default", + &file_url(&tmp_path.join("good.sql")), + STREAMING, + ) + .await?; + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + assert!(resp.body_string().await?.contains("already exists")); + Ok(()) + }); + + sim.run().unwrap(); +} diff --git a/libsql-server/tests/namespaces/mod.rs b/libsql-server/tests/namespaces/mod.rs index c33cc229c3..fa03c9b765 100644 --- a/libsql-server/tests/namespaces/mod.rs +++ b/libsql-server/tests/namespaces/mod.rs @@ -1,6 +1,7 @@ #![allow(deprecated)] mod dumps; +mod lifecycle; mod meta; mod shared_schema; @@ -20,6 +21,16 @@ fn make_primary(sim: &mut Sim, path: PathBuf) { } fn make_primary_with_db_config(sim: &mut Sim, path: PathBuf, db_config: DbConfig) { + make_primary_with(sim, path, db_config, false); +} + +/// A primary running in single-namespace mode (`--disable-namespaces`): only the default +/// namespace exists, and it is created at startup. +fn make_single_namespace_primary(sim: &mut Sim, path: PathBuf) { + make_primary_with(sim, path, DbConfig::default(), true); +} + +fn make_primary_with(sim: &mut Sim, path: PathBuf, db_config: DbConfig, single_namespace: bool) { init_tracing(); sim.host("primary", move || { let path = path.clone(); @@ -41,8 +52,8 @@ fn make_primary_with_db_config(sim: &mut Sim, path: PathBuf, db_config: DbConfig acceptor: TurmoilAcceptor::bind(([0, 0, 0, 0], 4567)).await?, tls_config: None, }), - disable_namespaces: false, - disable_default_namespace: true, + disable_namespaces: single_namespace, + disable_default_namespace: !single_namespace, ..Default::default() };