stardust.h

/* Existing: stardust.h, release alpha-2610112101. */

/* Subscribe copies the query's definition (frozen: later
 * definition edits do not apply) and parameters. Event 1 is the initial result.
 * Later events are full results computed after relevant commits.
 * sequence counts events delivered to this subscription.
 * superseded counts intermediate computed results this subscription skipped.
 * It does not count commits or changed rows.
 * The first event has superseded 0.
 * transaction is the database head used to compute the result. The caller owns event.result.
 * 
 * next waits up to max_wait_ticks (100-ns ticks). 0 polls. A
 * positive wait returns when an event exists. A negative wait is INVALID_ARGUMENT.
 * next returns OK, TIMEOUT, CANCELLED (after cancel or release), CLOSED (after
 * stardust_db_close) or the terminal failure, which subscription_error describes.
 * 
 * next calls can run concurrently. cancel can run on any thread. It wakes all
 * waiting next calls immediately and is terminal. A second cancel has no effect.
 * release cancels, waits for active calls and frees the subscription.
 */
/* Owned; released by stardust_db_subscription_release. */
typedef struct stardust_db_subscription stardust_db_subscription;
typedef struct {
	int64_t sequence;
	int64_t superseded;
	int64_t transaction;
	stardust_document *result; /* nullable */
} stardust_db_live_event;
/* errors Status, concurrency Shared */
uint32_t stardust_db_query_subscribe(stardust_db_query *query_handle, const stardust_document *parameters, const stardust_db_query_options *options, stardust_db_subscription **output);
/* errors Status, concurrency Shared */
uint32_t stardust_db_subscription_next(stardust_db_subscription *sub, int64_t max_wait_ticks, stardust_db_live_event *event);
/* errors Infallible, concurrency Shared */
void stardust_db_subscription_cancel(stardust_db_subscription *sub);
/* errors Infallible, concurrency Final, releases stardust_db_subscription */
void stardust_db_subscription_release(stardust_db_subscription *sub);

/* Proposed. */

/* set_notify lets a caller wait for a subscription without blocking a thread.
 * Call it once, after subscribe and before the first next. A second call is
 * INVALID_ARGUMENT.
 * Then call next with max_wait_ticks 0 until it returns TIMEOUT, and stop.
 * notify(ctx) runs once, on any thread, when next has something to return
 * again: an event, CANCELLED, CLOSED or a failure. Call next again then.
 * notify can also run when there is nothing, and next then returns TIMEOUT.
 * notify must not block or call stardust functions.
 * After release returns, notify is not running and does not run again.
 *
 *   subscribe(query, …)           -> sub
 *   set_notify(sub, notify, ctx)  once, for the life of the subscription
 *   next(sub, 0) -> OK            event 1, the initial result
 *   next(sub, 0) -> TIMEOUT       caught up; notify will run on the next change
 *   …                             wait: no thread is blocked
 *   notify(ctx)                   a commit changed the result
 *   next(sub, 0) -> OK            event 2
 *   next(sub, 0) -> TIMEOUT       caught up again
 *   …
 *   release(sub)                  notify never runs again
 */
/* errors Status, concurrency Exclusive */
uint32_t stardust_db_subscription_set_notify(stardust_db_subscription *sub, void (*notify)(void *ctx), void *ctx);

stardust.rs

// The public API: one consumer per subscription.

impl Subscription {
    /// The next event, or None once the subscription has ended
    /// (cancelled, database closed). Cancel-safe: dropping this future
    /// mid-wait loses no event.
    pub async fn next(&mut self) -> Result<Option<Event>>;

    /// A handle that can end the subscription from anywhere: another task,
    /// a signal handler, a Drop.
    pub fn canceller(&self) -> Canceller;
}

#[derive(Clone)]
pub struct Canceller(/* shares the native handle */);

impl Canceller {
    /// Ends the subscription; a pending next() wakes and returns None.
    pub fn cancel(&self);
}

// Dropping the Subscription cancels it.

// cancel moves to a separate Canceller because with &mut self you can't call
// sub.cancel() while another task is awaiting sub.next(). A cloneable handle
// fixes that, and it fits how cancellation is actually used: from a different
// place than the loop reading events. In our nu plugin, the Ctrl-C handler
// would hold a Canceller.

// Using it:

let mut sub = query.subscribe(None, &RunOptions::default())?;
let stop = sub.canceller();
on_ctrl_c(move || stop.cancel());

while let Some(event) = sub.next().await? {
    render(event.result);
}

// And the inside, on top of the set_notify ABI.

unsafe extern "C" {
    fn stardust_db_subscription_set_notify(
        sub: *mut native::DbSubscription,
        notify: extern "C" fn(ctx: *mut c_void),
        ctx: *mut c_void,
    ) -> u32;
}

// The native handle and the waker live together, shared by the Subscription
// and every Canceller. One waker, no list: there's only one consumer.
struct Inner {
    raw: NonNull<native::DbSubscription>,
    waker: AtomicWaker,
}

pub struct Subscription {
    inner: Arc<Inner>,
}

#[derive(Clone)]
pub struct Canceller(Arc<Inner>);

extern "C" fn notify(ctx: *mut c_void) {
    let inner = unsafe { &*(ctx as *const Inner) };
    inner.waker.wake();
}

impl Query {
    pub fn subscribe(&self, parameters: Option<&Document>, options: &RunOptions) -> Result<Subscription> {
        let raw = self.subscribe_raw(parameters, options)?; // stardust_db_query_subscribe
        let inner = Arc::new(Inner { raw, waker: AtomicWaker::new() });
        // Registered once, before anyone polls. The ABI arms it on each
        // next(0) that returns TIMEOUT. ctx is the Arc's heap address, stable
        // until the last owner drops.
        let status = unsafe {
            stardust_db_subscription_set_notify(
                inner.raw.as_ptr(),
                notify,
                Arc::as_ptr(&inner) as *mut c_void,
            )
        };
        check(status)?; // only fails if called twice, which this code never does
        Ok(Subscription { inner })
    }
}

impl Subscription {
    pub fn canceller(&self) -> Canceller {
        Canceller(Arc::clone(&self.inner))
    }
}

impl Canceller {
    pub fn cancel(&self) {
        // Any thread. Wakes a pending next(), which then returns None.
        unsafe { native::stardust_db_subscription_cancel(self.0.raw.as_ptr()) };
    }
}

impl Drop for Subscription {
    fn drop(&mut self) {
        // Dropping the Subscription cancels it. Cancellers may still exist,
        // so the handle itself is released later, by the last owner.
        unsafe { native::stardust_db_subscription_cancel(self.inner.raw.as_ptr()) };
    }
}

impl Drop for Inner {
    fn drop(&mut self) {
        // The last owner. Release cancels, waits for active calls and
        // guarantees notify is not running and never called again, so the
        // memory notify's ctx points at can be freed after this returns.
        unsafe { native::stardust_db_subscription_release(self.raw.as_ptr()) };
    }
}

impl Subscription {
    pub async fn next(&mut self) -> Result<Option<Event>> {
        std::future::poll_fn(|cx| {
            self.inner.waker.register(cx.waker()); // register first...
            match self.try_next() {                // ...then next(0), which arms on TIMEOUT
                Status::TIMEOUT => Poll::Pending,
                Status::OK => Poll::Ready(Ok(Some(self.take_event()))),
                Status::CANCELLED | Status::CLOSED => Poll::Ready(Ok(None)),
                failure => Poll::Ready(Err(self.error(failure))),
            }
        })
        .await
    }
}