Greptile Summary

This PR fixes a bug in DurabilityNotifier::spawn where the initial state was unconditionally set with closed_reason: None, causing any subscriber created after Db::close() to hang indefinitely rather than receiving an immediate error.

Key changes:

• initial_status (the full DbStatus snapshot) is now cloned from status_rx.borrow() instead of only reading durable_seq, allowing close_reason to be propagated into the initial State.
• tokio::spawn(run_notifier(...)) is now guarded by if initial_status.close_reason.is_none(), skipping the background task entirely when the DB is already closed — subscribers on a pre-closed notifier receive Err synchronously via the existing subscribe() fast-path.
• A regression test spawn_on_closed_db_fails_subscribers_immediately is added that builds a real slatedb::Db, closes it, creates a notifier, and asserts that the very first subscriber gets an immediate Err(CloseReason::Clean).

The fix is minimal, logically sound, and does not introduce any race conditions: the close_with_reason assertion (assert!(prev.is_none())) remains safe because run_notifier is only spawned when close_reason starts as None, ensuring close_with_reason is called at most once per notifier instance.

Confidence Score: 5/5

• This PR is safe to merge — it is a targeted bug fix with a focused regression test and no API surface changes.
• The change is small (one file, ~15 logical lines changed), well-reasoned, and comes with a concrete regression test against a real slatedb instance. The fix correctly handles the race between DB closure and notifier creation, the existing assert!(prev.is_none()) invariant in close_with_reason remains unviolated, and the subscribe() fast-path's priority order (closed_reason checked before durable_seq) ensures the test for subscribe(0, ...) on a closed DB yields Err as expected.
• No files require special attention.

Sequence Diagram

sequenceDiagram
    participant Caller
    participant DN as DurabilityNotifier
    participant Task as run_notifier task
    participant DB as slatedb watch channel

    Caller->>DN: spawn(db)
    DN->>DB: subscribe() returns status_rx
    DN->>DB: borrow() returns initial_status
    alt initial_status.close_reason is None
        DN->>Task: tokio::spawn(run_notifier)
        DN-->>Caller: Self { state: closed_reason=None }
        loop on each status change
            Task->>DB: changed().await
            Task->>Task: notify_waiters(durable_seq)
            opt close_reason is Some
                Task->>Task: close_with_reason then exit
            end
        end
    else initial_status.close_reason is Some
        Note over Task: Task NOT spawned
        DN-->>Caller: Self { state: closed_reason=Some }
    end

    Caller->>DN: subscribe(target_seq, callback)
    alt closed_reason is Some
        DN->>Caller: callback(Err(reason)) synchronously
    else last_durable_seq >= target_seq
        DN->>Caller: callback(Ok(durable_seq)) synchronously
    else pending
        DN->>DN: enqueue sorted Waiter
    end

_{Last reviewed commit: e4f731c}