Session with stream control sublanguage (#55)

SCL adds cursor-based streaming to Sakura so clients can consume DRL query results incrementally in bounded batches rather than all at once. Three statements: Begin, Fetch, and Close. `open`, `advance`, and `release` cursors backed by a session registry that suspends the generator between calls. SCL is just an optional consumption layer above the existing algebra.

7 files changed, 249 insertions, 1 deletions

diff --git a/bin/server.ml b/bin/server.ml
index 5a59e2e..2b9e651 100644
--- a/bin/server.ml
+++ b/bin/server.ml
@@ -99,8 +99,24 @@ let dbms_dispatch = Dispatch.create [
   (module Dml.Sublanguage.Memory : Dispatch.SubS);
   (module Icl.Sublanguage.Memory : Dispatch.SubS);
   (module Dcl.Sublanguage.Memory : Dispatch.SubS);
+  (module Scl.Sublanguage.Memory : Dispatch.SubS);
 ]
 
+let cursor_to_sexp storage db_name db_hash cursor_id rows has_more =
+  let open Sexplib.Sexp in
+  let row_sexps = List.map tuple_to_sexp rows in
+  let rows_sexp = List row_sexps in
+  to_string @@ List [
+    Atom "cursor";
+    List [Atom "id";        Atom cursor_id];
+    List [Atom "rows";      rows_sexp];
+    List [Atom "row_count"; Atom (string_of_int (List.length row_sexps))];
+    List [Atom "has_more";  Atom (string_of_bool has_more)];
+    List [Atom "db_hash";   Atom (short_hash db_hash)];
+    List [Atom "db_name";   Atom db_name];
+    List [Atom "branch";    Atom (get_branch storage)];
+  ]
+
 let execute_command storage db_ref cmd =
   let db   = !db_ref in
   let h    = db.Management.Database.hash in
@@ -114,6 +130,8 @@ let execute_command storage db_ref cmd =
     db_ref := new_db;
     advance_head_branch storage new_db.Management.Database.hash;
     ok new_db.Management.Database.hash msg
+  | Ok (Sublanguage.Cursor { cursor_id; rows; has_more }) ->
+    cursor_to_sexp storage name h cursor_id rows has_more
   | Error e ->
     err h (Dispatch.sexp_of_dispatch_error e)
 
@@ -223,6 +241,9 @@ let warn_on_error fmt = function
   | Ok ()   -> ()
   | Error e -> Printf.eprintf fmt e
 
+let cursor_gc_max_age  = 300.0  (* 5 minutes *)
+let cursor_gc_interval = 60.0   (* run GC at most once per minute *)
+
 let () =
   match setup () with
   | Error msg -> failwith msg
@@ -231,12 +252,16 @@ let () =
       |> register_prelude_relations storage
       |> register_branch_relations storage
     in
+    (* Initialise cursor session registry for SCL *)
+    let session_reg = Session.create () in
+    Scl.Executor.set_sessions session_reg;
     (* Create default master branch in raw registry and set HEAD *)
     BranchOps.create storage ~name:"master" ~tip:db.Management.Database.hash
     |> warn_on_error "Warning: failed to create master branch: %s\n%!";
     BranchOps.checkout storage "master"
     |> warn_on_error "Warning: failed to checkout master: %s\n%!";
     let db_ref = ref db in
+    let last_gc = ref 0.0 in
     let sock = Unix.socket Unix.PF_INET Unix.SOCK_STREAM 0 in
     Unix.setsockopt sock Unix.SO_REUSEADDR true;
     Unix.setsockopt sock Unix.SO_REUSEPORT true;
@@ -247,5 +272,10 @@ let () =
     while true do
       let (client_fd, _addr) = Unix.accept sock in
       handle_client storage db_ref client_fd;
-      (try Unix.close client_fd with _ -> ())
+      (try Unix.close client_fd with _ -> ());
+      let now = Unix.gettimeofday () in
+      if now -. !last_gc >= cursor_gc_interval then begin
+        Session.gc session_reg ~max_age:cursor_gc_max_age;
+        last_gc := now
+      end
     done
diff --git a/lib/scl/ast.ml b/lib/scl/ast.ml
new file mode 100644
index 0000000..043f544
--- /dev/null
+++ b/lib/scl/ast.ml
@@ -0,0 +1,7 @@
+open Sexplib.Std
+
+type statement =
+  | Begin of { query : Drl.Ast.query; limit : int option }
+  | Fetch of { cursor : string; limit : int option }
+  | Close of { cursor : string }
+[@@deriving sexp]
diff --git a/lib/scl/executor.ml b/lib/scl/executor.ml
new file mode 100644
index 0000000..6cb17b8
--- /dev/null
+++ b/lib/scl/executor.ml
@@ -0,0 +1,77 @@
+let default_batch = 50
+
+type cursor_batch = {
+  cursor_id : string;
+  rows : Tuple.materialized list;
+  has_more : bool;
+}
+
+type exec_result =
+  | Batch of cursor_batch
+  | Closed of Management.Database.t
+
+let sessions : Session.t option ref = ref None
+
+let set_sessions s = sessions := Some s
+
+module Make (Storage : Management.Physical.S) = struct
+  module DrlExec = Drl.Executor.Make(Storage)
+  module Alg = Algebra.Make(Storage)
+
+  type error =
+    | ParseError  of string
+    | QueryError  of DrlExec.error
+    | CursorError of string
+
+  let sexp_of_error e =
+    let open Sexplib.Sexp in
+    match e with
+    | ParseError s  -> List [Atom "parse-error";  Atom s]
+    | QueryError e  -> DrlExec.sexp_of_error e
+    | CursorError s -> List [Atom "cursor-error"; Atom s]
+
+  let get_sessions () =
+    match !sessions with
+    | Some s -> Ok s
+    | None   -> Error (CursorError "The session registry has not been initialised; ensure `set_sessions` is called at startup.")
+
+  let ( let* ) = Result.bind
+
+  let execute
+      (storage : Storage.t)
+      (db : Management.Database.t)
+      (stmt : Ast.statement)
+    : (exec_result, error) result =
+    match stmt with
+    | Ast.Begin { query; limit } ->
+      let* reg = get_sessions () in
+      let* rel =
+        DrlExec.execute storage db query
+        |> Result.map_error (fun e -> QueryError e)
+      in
+      let gen = Alg.to_generator storage rel in
+      let query_str = Drl.Parser.to_string query in
+      let cur = Session.register reg ~db ~query:query_str ~generator:gen in
+      let batch = Option.value ~default:default_batch limit in
+      let* (rows, has_more) =
+        Session.fetch reg ~id:cur.Session.id ~limit:batch
+        |> Result.map_error (fun s -> CursorError s)
+      in
+      Ok (Batch { cursor_id = cur.Session.id; rows; has_more })
+
+    | Ast.Fetch { cursor; limit } ->
+      let* reg = get_sessions () in
+      let batch = Option.value ~default:default_batch limit in
+      let* (rows, has_more) =
+        Session.fetch reg ~id:cursor ~limit:batch
+        |> Result.map_error (fun s -> CursorError s)
+      in
+      Ok (Batch { cursor_id = cursor; rows; has_more })
+
+    | Ast.Close { cursor } ->
+      let* reg = get_sessions () in
+      Session.close reg ~id:cursor;
+      Ok (Closed db)
+end
+
+module Memory = Make(Management.Physical.Memory)
diff --git a/lib/scl/parser.ml b/lib/scl/parser.ml
new file mode 100644
index 0000000..3e9c4ed
--- /dev/null
+++ b/lib/scl/parser.ml
@@ -0,0 +1,19 @@
+type error = ParseError of string
+
+let validate = function
+  | (Ast.Begin { limit = Some limit; _ } | Ast.Fetch { limit = Some limit; _ })
+    when limit <= 0 ->
+      Error (ParseError "The provision of `limit` is expected to be positive.")
+  | stmt -> Ok stmt
+
+let of_sexp sexp =
+  match Ast.statement_of_sexp sexp with
+  | exception exn -> Error (ParseError (Printexc.to_string exn))
+  | stmt -> validate stmt
+
+let of_string s =
+  match Sexplib.Sexp.of_string s |> Ast.statement_of_sexp with
+  | exception exn -> Error (ParseError (Printexc.to_string exn))
+  | stmt -> validate stmt
+
+let to_string stmt = Ast.sexp_of_statement stmt |> Sexplib.Sexp.to_string_hum
diff --git a/lib/scl/sublanguage.ml b/lib/scl/sublanguage.ml
new file mode 100644
index 0000000..7098ced
--- /dev/null
+++ b/lib/scl/sublanguage.ml
@@ -0,0 +1,26 @@
+module Make (Storage : Management.Physical.S) = struct
+  module Exec = Executor.Make(Storage)
+
+  type storage = Storage.t
+  type ast     = Ast.statement
+  type error   = Exec.error
+
+  let name = "scl"
+
+  let parse_sexp sexp =
+    match Parser.of_sexp sexp with
+    | Ok r -> Ok r
+    | Error (Parser.ParseError s) -> Error (Exec.ParseError s)
+
+  let execute storage db ast =
+    Exec.execute storage db ast
+    |> Result.map (function
+      | Executor.Batch { cursor_id; rows; has_more } ->
+        Sublanguage.Cursor { cursor_id; rows; has_more }
+      | Executor.Closed db ->
+        Sublanguage.Transition (db, "cursor closed"))
+
+  let sexp_of_error = Exec.sexp_of_error
+end
+
+module Memory = Make(Management.Physical.Memory)
diff --git a/lib/session.ml b/lib/session.ml
new file mode 100644
index 0000000..272a419
--- /dev/null
+++ b/lib/session.ml
@@ -0,0 +1,82 @@
+(** Cursor registry for streaming tuple consumption.
+
+    Each cursor wraps a [Generator.t] produced by a DRL query and
+    tracks the current position so that successive [fetch] calls
+    return the next batch of rows without replaying the query. *)
+
+type cursor = {
+  id : string;
+  mutable generator : Generator.t;
+  mutable position : int;
+  db_snapshot : Management.Database.t;
+  query_sexp : string;
+  created_at : float;
+}
+
+type t = {
+  tbl : (string, cursor) Hashtbl.t;
+  mutable counter : int;
+}
+
+let create () : t =
+  { tbl = Hashtbl.create 16; counter = 0 }
+
+let register (reg : t) ~(db : Management.Database.t) ~(query : string)
+    ~(generator : Generator.t) : cursor =
+  let raw = string_of_int reg.counter ^ query ^ db.Management.Database.hash in
+  let id = Conventions.Hash.hash_text raw in
+  reg.counter <- reg.counter + 1;
+  let cur = {
+    id;
+    generator;
+    position = 0;
+    db_snapshot = db;
+    query_sexp = query;
+    created_at = Unix.gettimeofday ();
+  } in
+  Hashtbl.replace reg.tbl id cur;
+  cur
+
+let fetch (reg : t) ~(id : string) ~(limit : int)
+    : (Tuple.materialized list * bool, string) result =
+  match Hashtbl.find_opt reg.tbl id with
+  | None -> Error ("The cursor with identifier `" ^ id ^ "` was not found in the session registry.")
+  | Some cur ->
+    let rec go gen pos acc count =
+      if count >= limit then (List.rev acc, gen, pos, true)
+      else
+        match gen (Some pos) with
+        | Generator.Done    -> (List.rev acc, gen, pos, false)
+        (* Generator errors are treated as exhaustion: the partial batch is
+           returned with has_more = false and the cursor is auto-removed.
+           The caller receives no indication that the stream terminated
+           abnormally. When fetch gains a richer return type this arm
+           should surface the error message instead. *)
+        | Generator.Error _ -> (List.rev acc, gen, pos, false)
+        | Generator.Value (t, next) ->
+          begin match t with
+          | Tuple.Materialized m ->
+            go next (pos + 1) (m :: acc) (count + 1)
+          | Tuple.NonMaterialized _ ->
+            go next (pos + 1) acc count
+          end
+    in
+    let (rows, gen', pos', has_more) =
+      go cur.generator cur.position [] 0
+    in
+    cur.generator <- gen';
+    cur.position <- pos';
+    if not has_more then Hashtbl.remove reg.tbl id;
+    Ok (rows, has_more)
+
+let close (reg : t) ~(id : string) : unit =
+  Hashtbl.remove reg.tbl id
+
+let gc (reg : t) ~(max_age : float) : unit =
+  let now = Unix.gettimeofday () in
+  let to_remove =
+    Hashtbl.fold (fun k cur acc ->
+      if now -. cur.created_at >= max_age then k :: acc else acc)
+      reg.tbl []
+  in
+  List.iter (Hashtbl.remove reg.tbl) to_remove
diff --git a/lib/sublanguage.ml b/lib/sublanguage.ml
index 3df91f1..d92358e 100644
--- a/lib/sublanguage.ml
+++ b/lib/sublanguage.ml
@@ -14,9 +14,16 @@
       (Turing-complete; breaks the termination guarantee of DRL)
     - ACL -> a functor restriction: narrows the admissible natural
       transformations based on authorization context *)
+type cursor_result = {
+  cursor_id : string;
+  rows : Tuple.materialized list;
+  has_more : bool;
+}
+
 type result =
   | Query of Relation.t
   | Transition of Management.Database.t * string
+  | Cursor of cursor_result
 
 module type S = sig
   type storage