Possible deadlock identified in DatabaseInner::drop()

[jrobhoward]

Describe the bug
DatabaseInner::drop() shuts down workers by sending Close messages through the same bounded work channel that carries Flush/Compact/RotateMemtable messages. If the channel is full (which can happen when writers are still active during drop), the blocking send(Close) can deadlock. Workers can't drain the channel because they themselves block on send() & nobody makes progress.

To Reproduce
This (extracted/simplified?) example can usually reproduce a deadlock (cargo run --release) in under 25 iterations on my machine:

[package]
name = "fjall-deadlock-repro"
version = "0.1.0"
edition = "2021"

[dependencies]
fjall = "3"
tempfile = "3"

use fjall::{Database, KeyspaceCreateOptions};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};
use tempfile::TempDir;

/// Reproducer for fjall 3.x deadlock-on-drop.
///
/// The deadlock occurs in DatabaseInner::drop() (db.rs:74-81) which uses
/// blocking `sender.send(WorkerMessage::Close)` on a bounded(1000) flume
/// channel. Two interlocking problems cause the deadlock:
///
/// 1. FIFO starvation: When writer threads are still active during Drop,
///    they pump RotateMemtable messages via try_send(). Drop's drain()
///    empties the channel, but writers immediately refill it. Close messages
///    enter the channel BEHIND the writer-generated messages. Workers dequeue
///    and process those messages first, generating more Flush/Compact messages,
///    pushing Close further back.
///
/// 2. Mutual blocking: Workers themselves use blocking send() on the same
///    bounded channel — worker #0 re-sends Compact (worker_pool.rs:224) and
///    inner_rotate_memtable sends Flush (keyspace/mod.rs:749). When the
///    channel is full, these workers block and can't recv ANY messages,
///    including Close. Meanwhile Drop blocks on send(Close) for the same
///    reason. Nobody can make progress.
///
/// Key conditions:
///   1. Writers must still be active when DatabaseInner::drop() runs
///   2. Tiny memtable size forces rotation on nearly every insert
///   3. Multiple keyspaces multiply the message traffic
fn attempt() {
    let dir = TempDir::new().unwrap();
    let stop = Arc::new(AtomicBool::new(false));

    // Use default worker threads (pool_size > 1 enables worker #0 blocking re-send)
    let db = Database::builder(dir.path()).open().unwrap();

    // Tiny memtable (1KB) forces rotation on nearly every insert.
    // Each rotation → RotateMemtable msg → Flush msg → pool_size × Compact msgs.
    let small_memtable = || KeyspaceCreateOptions::default().max_memtable_size(1_000);

    let ks1 = db.keyspace("ks1", small_memtable).unwrap();
    let ks2 = db.keyspace("ks2", small_memtable).unwrap();
    let ks3 = db.keyspace("ks3", small_memtable).unwrap();
    let ks4 = db.keyspace("ks4", small_memtable).unwrap();

    // Spawn writer threads that hold Keyspace clones and write continuously.
    // These threads keep running AFTER we drop the Database below.
    let mut writer_handles = vec![];
    for t in 0..16u32 {
        let ks = match t % 4 {
            0 => ks1.clone(),
            1 => ks2.clone(),
            2 => ks3.clone(),
            _ => ks4.clone(),
        };
        let stop = stop.clone();
        writer_handles.push(thread::spawn(move || {
            let mut k = 0u64;
            while !stop.load(Ordering::Relaxed) {
                let key = format!("t{t:02}_k{k:012}");
                // Errors are expected once the database starts dropping
                if ks.insert(key.as_bytes(), vec![0xAB; 256]).is_err() {
                    break;
                }
                k += 1;
            }
        }));
    }

    // Let writers build up pressure — the channel should be near capacity
    // with RotateMemtable/Flush/Compact messages from the tiny memtable.
    thread::sleep(Duration::from_millis(200));

    // Drop the Database while writers are still active.
    // This triggers DatabaseInner::drop() which tries to send Close messages,
    // but writers keep flooding the channel with RotateMemtable messages.
    //
    // IMPORTANT: We drop `db` but NOT the keyspace clones held by writers.
    // The Keyspace holds a flume::Sender clone (worker_messager) so writers
    // can still try_send into the channel even after Drop starts.
    drop(ks1);
    drop(ks2);
    drop(ks3);
    drop(ks4);
    drop(db); // <-- This triggers DatabaseInner::drop(), which may deadlock

    // Signal writers to stop (if Drop completed without deadlock)
    stop.store(true, Ordering::Relaxed);
    for h in writer_handles {
        h.join().ok();
    }
}

fn main() {
    let timeout = Duration::from_secs(10);
    let num_iterations = 200;

    eprintln!("fjall 3.x deadlock-on-drop reproducer");
    eprintln!(
        "{} iterations, {}s timeout each\n",
        num_iterations,
        timeout.as_secs()
    );

    for i in 0..num_iterations {
        eprint!("\rIteration {}/{}", i + 1, num_iterations);

        let start = Instant::now();
        let handle = thread::spawn(attempt);

        loop {
            if handle.is_finished() {
                handle.join().unwrap();
                break;
            }
            if start.elapsed() > timeout {
                eprintln!(
                    "\n\nDEADLOCK DETECTED at iteration {}! (hung for {:?})",
                    i + 1,
                    timeout
                );
                eprintln!("DatabaseInner::drop() blocked on flume::send(WorkerMessage::Close)");
                eprintln!("\nRoot cause: two interlocking problems on the bounded(1000) channel:");
                eprintln!(
                    "  1. Writers pump RotateMemtable messages, starving Close in the FIFO queue"
                );
                eprintln!(
                    "  2. Workers block on send() (Compact re-send, Flush in rotate_memtable),"
                );
                eprintln!("     so they can't recv Close even if it reaches the front");
                std::process::exit(1);
            }
            thread::sleep(Duration::from_millis(50));
        }
    }

    eprintln!(
        "\n\nCompleted {} iterations without deadlock.",
        num_iterations
    );
    eprintln!("Try adjusting parameters or use `cargo bench`.");
}

Additional context
I don't normally use the DB this way, it's just something I found during a concurrent benchmark test. I'm unsure how likely it is to happen in a normal/production workload. This may be related to existing bugs #257 and/or #183 .

The fix may be to not use the work channel for shutdown. The existing stop_signal (AtomicBool) is already set in Drop but never checked by workers. Workers could probably check it directly and exit.