Watching a statement while it runs

README.md

@@ -75,7 +75,7 @@ The switches come across, including `bigIntMode` and `temporal`, because a pool
 
 ## What works today
 
-`connect`, `query`, `exec`, `stream`, `close`, `dispose` and `await using`. `duplicate`, for a second connection made from the first. Named parameters both ways, including lists, records and nesting. Every scalar the engine has, plus nodes, edges and paths with their tables named rather than numbered, and `ZuDate`, `ZuTime`, `ZuTimestamp` and `ZuDuration`, with `{ temporal: true }` and `toTemporal()` for the runtimes that have `Temporal`. Read-only connections, databases in memory, memory and thread limits. `bigIntMode`, per statement or per connection. An `AbortSignal` on any statement. The full error surface above, and `isZuError` to recognize it. Streaming, as an async iterable, as batches and as a Web Stream. Transactions, with `inTransaction` on the connection. An appender, for loading rows a batch at a time, and `load` for building a whole database out of columns and an edge list. Registered frames, so an Arrow table or an object of typed arrays is something a statement can match on without the rows being copied. `columnar`, for a result read down its columns as the buffers themselves rather than across its rows as objects. Prepared statements, compiled at the line that asked and run as often as wanted, and `explain` and `profile`, as a tree a program walks and as the listing a person reads. Both module formats, typed separately.
+`connect`, `query`, `exec`, `stream`, `close`, `dispose` and `await using`. `duplicate`, for a second connection made from the first. Named parameters both ways, including lists, records and nesting. Every scalar the engine has, plus nodes, edges and paths with their tables named rather than numbered, and `ZuDate`, `ZuTime`, `ZuTimestamp` and `ZuDuration`, with `{ temporal: true }` and `toTemporal()` for the runtimes that have `Temporal`. Read-only connections, databases in memory, memory and thread limits. `bigIntMode`, per statement or per connection. An `AbortSignal` on any statement, and `rowsRead` and `progress` for watching the one running now. The full error surface above, and `isZuError` to recognize it. Streaming, as an async iterable, as batches and as a Web Stream. Transactions, with `inTransaction` on the connection. An appender, for loading rows a batch at a time, and `load` for building a whole database out of columns and an edge list. Registered frames, so an Arrow table or an object of typed arrays is something a statement can match on without the rows being copied. `columnar`, for a result read down its columns as the buffers themselves rather than across its rows as objects. Prepared statements, compiled at the line that asked and run as often as wanted, and `explain` and `profile`, as a tree a program walks and as the listing a person reads. Both module formats, typed separately.
 
 Build it with `npm run build`, and run the suite with `npm test`. Nothing is published yet, so `npm i zudb` is not a thing you can type at anybody's terminal, but everything it will do is built and installed on every run of the release workflow.
 
@@ -91,6 +91,19 @@ It is the signal JavaScript already has, so a timeout written like the one above
 
 What the promise rejects with is the signal's own reason, which is what `fetch` does: `AbortSignal.timeout(50)` rejects with the runtime's `TimeoutError`, `controller.abort(new RequestGone())` rejects with the `RequestGone` you made, and a bare `controller.abort()` rejects with the runtime's `AbortError`. A signal that has already fired stops the statement before the engine sees it at all. A signal that never fires costs one listener, taken off again when the statement ends, whether it answered, failed or was stopped.
 
+## Watching one run
+
+A statement that takes a minute is one somebody is sitting in front of, so a connection says how far the one running now has got:
+
+```ts
+using watch = conn.progress((rows) => process.stdout.write(`\r${rows} rows read`));
+const answer = await conn.query(statement);
+```
+
+`rowsRead` is the number underneath it, and it is a property rather than a call because reading it must never wait: the statement is on a threadpool thread holding the connection's lock, and this is an atomic beside the lock rather than a question through it. It counts rows read out of storage rather than rows answered, because the statement somebody is waiting on is exactly the one that reads a hundred million rows to answer one. It starts again at zero at each statement and holds its last value once one ends, so `conn.rowsRead` after a statement is what that statement cost.
+
+`progress` is a timer around that number, a tenth of a second apart unless you say otherwise with `{ everyMs }`. The callback runs only when the count has moved, which is what makes a watch on an idle connection quiet, and the timer does not hold the event loop open, so a watch nobody stopped is not a program that never exits. Stop it with `stop()` or by leaving the scope of the `using`. Nothing calls into JavaScript from the thread doing the scanning, which is the point: the statement being watched does not know it is being watched and does not slow down for it.
+
 ## Reading a result a piece at a time
 
 `conn.stream(...)` runs the same statement and hands the rows over as they are made, instead of building the whole answer first:

binding.d.cts

@@ -569,6 +569,33 @@ export interface ZuProfile {
   readonly text: string
 }
 
+/**
+ * What a watch on a running statement takes.
+ */
+export interface ZuProgressOptions {
+  /**
+   * How long to wait between looks, in milliseconds. A tenth of a
+   * second by default, which is about where a person stops reading a
+   * number and starts seeing it move.
+   *
+   * What one look costs is an atomic read, so this is a question about
+   * how often the callback should run rather than about how much the
+   * watch costs the statement.
+   */
+  readonly everyMs?: number
+}
+
+/**
+ * A watch on a running statement, which is stopped by `stop()` or by
+ * leaving the scope of a `using`.
+ *
+ * Stopping twice does nothing, and so does stopping one that has
+ * already been left behind.
+ */
+export interface ZuProgress extends Disposable {
+  stop(): void
+}
+
 /**
  * What a streamed statement takes beside its parameters.
  */
@@ -827,6 +854,28 @@ export declare class Connection {
    * atomic.
    */
   get inTransaction(): boolean
+  /**
+   * How many rows the statement running on this connection has read
+   * out of storage, for showing a person that something is
+   * happening.
+   *
+   * Rows read rather than rows answered, because the statement
+   * somebody is waiting on is exactly the one that reads a hundred
+   * million rows to answer one. It starts at zero at each statement
+   * and holds its last value once one ends.
+   *
+   * This is the one thing on a connection that is worth reading
+   * while a statement runs, and it is answered the way
+   * [`Connection::open`] is: an atomic beside the lock rather than a
+   * question through it. So the loop's thread gets its answer while
+   * the threadpool thread is still scanning, and `progress()` is the
+   * timer written around it.
+   *
+   * A number rather than a bigint, like every other count this
+   * client makes rather than reads out of a column: a statement that
+   * had read 2^53 rows would have been running for weeks.
+   */
+  get rowsRead(): number
   /**
    * Starts a transaction and hands it back.
    *

etc/zudb.api.md

@@ -44,6 +44,7 @@ export class Connection {
     get readOnly(): boolean
     register(name: string, data: ZuFrame): Promise<number>
     registered(): Promise<string[]>
+    get rowsRead(): number
     transaction(options?: ZuTransactionOptions | null): Promise<Transaction>
     unregister(name: string): Promise<void>
 }
@@ -370,6 +371,17 @@ export interface ZuProfile {
     readonly text: string
 }
 
+// @public
+export interface ZuProgress extends Disposable {
+    // (undocumented)
+    stop(): void
+}
+
+// @public
+export interface ZuProgressOptions {
+    readonly everyMs?: number
+}
+
 // @public
 export class ZuRel {
     // (undocumented)

src/conn.rs

@@ -479,6 +479,30 @@ impl Connection {
         self.in_txn.load(Ordering::Acquire)
     }
 
+    /// How many rows the statement running on this connection has read
+    /// out of storage, for showing a person that something is
+    /// happening.
+    ///
+    /// Rows read rather than rows answered, because the statement
+    /// somebody is waiting on is exactly the one that reads a hundred
+    /// million rows to answer one. It starts at zero at each statement
+    /// and holds its last value once one ends.
+    ///
+    /// This is the one thing on a connection that is worth reading
+    /// while a statement runs, and it is answered the way
+    /// [`Connection::open`] is: an atomic beside the lock rather than a
+    /// question through it. So the loop's thread gets its answer while
+    /// the threadpool thread is still scanning, and `progress()` is the
+    /// timer written around it.
+    ///
+    /// A number rather than a bigint, like every other count this
+    /// client makes rather than reads out of a column: a statement that
+    /// had read 2^53 rows would have been running for weeks.
+    #[napi(getter)]
+    pub fn rows_read(&self) -> f64 {
+        self.interrupt.rows() as f64
+    }
+
     /// Starts a transaction and hands it back.
     ///
     /// It starts here rather than at the first statement inside it, so a
@@ -1150,6 +1174,24 @@ pub(crate) fn with<T>(
     answered
 }
 
+/// A statement is about to run, and the counter it reports its rows
+/// through starts again at zero.
+///
+/// Called where the connection becomes one statement's, which is the
+/// only moment the count can be reset without racing the statement
+/// reading it: the lock is held here and the reader is a getter that
+/// takes no lock at all. Held rather than cleared afterwards, so that
+/// `rowsRead` after a statement is what that statement cost.
+///
+/// The word an interrupt is raised through is put down by the same
+/// call, which is the reason this happens before the signal is entered
+/// rather than after: a signal that fired while nothing was running
+/// raised nothing to put down, and one that fires from here on is
+/// answered by the watch instead.
+pub(crate) fn began(conn: &mut zudb::Connection) {
+    conn.interrupt().clear();
+}
+
 /// What a closed connection says, wherever it is noticed.
 pub(crate) const CLOSED: &str =
     "the connection is closed, so there is nothing left to run a statement on";
@@ -1470,6 +1512,7 @@ impl QueryTask {
             // being able to. A signal that fired first ends the
             // statement without the engine ever seeing it, which is the
             // whole point of asking.
+            began(conn);
             if let Some(watch) = watch
                 && !watch.enter()
             {

src/plan.rs

@@ -47,7 +47,7 @@ use zudb::query::Value;
 use zudb::{OpProfile, PlanNode, Profile, QueryPlan, StageProfile, ZuError};
 
 use crate::cancel::Watch;
-use crate::conn::{Failure, Handles, failed, with};
+use crate::conn::{Failure, Handles, began, failed, with};
 
 /// Compiling a statement to see what it would do.
 pub struct PlanTask {
@@ -136,7 +136,9 @@ impl<'task> ScopedTask<'task> for ProfileTask {
             move |conn| {
                 // A profile is a run, so it is stoppable exactly the way
                 // a run is: the signal is entered when the connection
-                // becomes this call's and left when it stops being.
+                // becomes this call's and left when it stops being, and
+                // the row counter starts again the way it does for one.
+                began(conn);
                 if let Some(watch) = watch
                     && !watch.enter()
                 {

src/stream.rs

@@ -34,7 +34,7 @@ use zudb::query::Value;
 use zudb::{Batch, Flow, Streamed, ZuError};
 
 use crate::cancel::{Guard, Watch};
-use crate::conn::{CLOSED, Failure, STREAMING, beside, failed, notices};
+use crate::conn::{CLOSED, Failure, STREAMING, began, beside, failed, notices};
 use crate::value::{Shape, Spelling, to_js};
 
 /// How many batches may sit between the statement and the reader.
@@ -674,6 +674,7 @@ impl Started {
         // From here the connection is this statement's, so this is
         // where a signal can start stopping it. A signal that fired
         // first ends the statement without the engine ever seeing it.
+        began(conn);
         if let Some(guard) = &self.guard
             && !guard.enter()
         {