Module Neodriver_eio.Conn

A minimal Bolt connection (connect, authenticate, RUN/PULL/DISCARD, transactions).

Minimal Bolt connection: TCP connect (+ optional TLS) + handshake + HELLO/auth + state machine.

See conn.ml for the implementation.

Authentication token sent in HELLO (Bolt <= 5.0) or LOGON (Bolt >= 5.1): the basic scheme by default, but bearer and custom schemes are supported too (absent fields are omitted on the wire).

type config = {
  1. host : string;
  2. port : int;
  3. scheme : Neodriver_core.Addressing.scheme;
  4. connection_timeout : float;
  5. user_agent : string;
  6. auth : auth;
  7. routing_context : (string * string) list option;
  8. telemetry_disabled : bool;
  9. notifications_min_severity : string option;
  10. notifications_disabled_categories : string list option;
}

Target connection settings. The scheme selects TLS: Bolt plain, Bolt_secure TLS with certificate validation, Bolt_self_signed TLS without validation. routing_context is sent as the routing field of HELLO (server-side routing) for routed (neo4j*) drivers: None for direct bolt* drivers, Some ctx for routed ones (an empty list sends routing: {}). The field is only sent on Bolt >= 4.1. telemetry_disabled suppresses the Bolt 5.4+ TELEMETRY notifications. notifications_min_severity and notifications_disabled_categories are the driver-level notification filtering settings carried in HELLO on Bolt >= 5.2 (None omits the field; Some [] sends an empty category list).

type t

An established, authenticated Bolt connection: the transport, the negotiated protocol version and the tracked server state.

val default_user_agent : string

Default user_agent header for HELLO.

val basic_auth : ?principal:string -> ?credentials:string -> ?realm:string -> unit -> auth

The basic authentication token (scheme = "basic"), with the given principal (default neo4j), credentials (default empty) and optional realm.

val bearer_auth : string -> auth

A bearer (SSO) token (scheme = "bearer"), with the token as credentials.

val connect : ?resolver: (Neodriver_core.Addressing.t -> (Neodriver_core.Addressing.t list, Neodriver_core.Errors.t) Stdlib.result) -> ?domain_name_resolver: (string -> (string list, Neodriver_core.Errors.t) Stdlib.result) -> [> `Network | `Platform of [> `Generic ] ] Eio.Resource.t -> Mtime.t Eio.Time.clock_ty Eio.Resource.t -> Eio.Switch.t -> config -> (t, Neodriver_core.Errors.t) Stdlib.result

Establish a connection (over TLS when the scheme requires it), negotiate the Bolt protocol version and authenticate. resolver replaces the address lookup: the address built from config is passed to it and each returned address is tried in turn (first success wins, errors are aggregated). Without resolver, the single configured address is used. domain_name_resolver resolves any hostnames among the (resolved) addresses to literal IPs (the TestKit harness's custom domain-name resolution); literal IPs are used as-is. An IPv6 literal in config.host is treated as such (the address is built with brackets around the host). For Bolt >= 5.1 the authentication is sent via LOGON after HELLO; for older versions it is inline in HELLO. clock and config.connection_timeout bound the TCP connect / TLS handshake and the connection's subsequent reads and writes. The Bolt version handshake itself is not bounded by config.connection_timeout: it is expected to run inside a connection-acquisition deadline that bounds the whole connect (the pool's / routing cluster's acquisition timeout fires first when it is smaller, and a handshake that outlives the socket connection timeout but fits in the acquisition deadline still succeeds).

  • returns

    Error _ for connection/handshake failures, for routing schemes (unsupported until routing is implemented), or for an authentication failure reported by the server.

The resolved address the connection is established with.

val version : t -> int * int

The negotiated protocol version (major, minor).

val id : t -> int

The connection id (a driver-assigned counter, rendered as "#XXXX" in log lines; see Log.conn).

val server_state : t -> State.t

The tracked server protocol state.

val is_failed : t -> bool

Whether the server answered the last request with a FAILURE: the connection is not in a clean state and needs a RESET before it can be reused.

val set_on_error : t -> (t -> Neodriver_core.Errors.t -> Neodriver_core.Errors.t) -> unit

Install a callback invoked with t and the error whenever a request on the connection fails: failed auto-RESETs and failed messages in run and route, and server failures surfaced by pull and discard. The callback returns the error to surface to the caller — an auth manager or address-deactivation hook may return a modified error (e.g. one marked retryable). The routing cluster installs it to deactivate the connection's address.

The currently installed on-error callback (the identity by default). Callers that install an additional hook should compose it with the existing one (e.g. the pool chains its security-error handling after the cluster's address deactivation).

val set_auth_manager : t -> Neodriver_core.Auth_manager.t -> unit

Install the auth manager behind the connection's current token. The pool installs it so the set_on_error hook can call handle_security_exception when the server returns a security error (see auth_manager).

val auth_manager : t -> Neodriver_core.Auth_manager.t option

The connection's auth manager, if one was installed.

val last_database : t -> string option

The db of the last run on the connection, if any.

val set_last_database : t -> string option -> unit

Record the db of the next run.

val reset : t -> (unit, Neodriver_core.Errors.t) Stdlib.result

Send a RESET and wait for the response; the server returns to Ready.

val logon : t -> auth -> (unit, Neodriver_core.Errors.t) Stdlib.result

Re-authenticate with auth via LOGON (Bolt >= 5.1 only). A RESET is sent first if the server is in the Failed state.

  • returns

    Error _ for older protocol versions or on server failure.

val logoff : t -> (unit, Neodriver_core.Errors.t) Stdlib.result

De-authenticate via LOGOFF (Bolt >= 5.1 only). A RESET is sent first if the server is in the Failed state.

  • returns

    Error _ for older protocol versions or on server failure.

val close : t -> unit

Close the connection.

val hydration : t -> Neodriver_core.Hydration.t

A fresh hydration scope for the connection's protocol version.

type run_metadata = {
  1. fields : string list;
  2. qid : int option;
  3. bookmark : string option;
  4. t_first : int option;
  5. rt : Neodriver_packstream.Packstream.value option;
  6. db : string option;
}

Metadata of a RUN response: the result's field names, the query id (for multiple results), the bookmark reported for an auto-commit transaction (if any), the t_first timing (result available-after, milliseconds), the rt routing-table value reported when server-side routing is enabled (if any), and the db the server reports it actually used for the query (if any).

val run : ?mode:Neodriver_core.Config.access_mode -> ?db:string -> ?imp_user:string -> ?bookmarks:string list -> ?timeout:float -> ?metadata:(string * Neodriver_core.Values.t) list -> ?telemetry:int -> ?notifications_min_severity:string -> ?notifications_disabled_categories:string list -> ?fetch_size:int -> t -> hydration:Neodriver_core.Hydration.t -> query:string -> parameters:(string * Neodriver_core.Values.t) list -> (run_metadata, Neodriver_core.Errors.t) Stdlib.result

Send a RUN message for query. parameters are dehydrated with hydration. The optional mode, db, bookmarks, timeout (seconds) and metadata (tx_metadata) go into the request's extra map. telemetry batches a TELEMETRY notification (Bolt 5.4+) with the RUN. notifications_min_severity and notifications_disabled_categories are the session-level notification filtering settings, sent in the extra map on Bolt >= 5.2. The first PULL is pipelined with the RUN (fetch_size records, default 1000; Bolt 3 sends PULL_ALL): its response is consumed by the result's pull/discard (or drain_pending_pull when abandoned).

  • returns

    Error _ if the server fails the request (the connection enters Failed and is RESET before the next request).

val begin_ : ?telemetry:int -> t -> extra:Neodriver_packstream.Packstream.value -> (string option, Neodriver_core.Errors.t) Stdlib.result

Send a BEGIN message (start a transaction) with the given extra map (see build_extra). A RESET is sent first if the server is in the Failed state. telemetry batches a TELEMETRY notification with the BEGIN. Returns the db the server reports the transaction runs on (Bolt 5.2+), if any.

val begin_pipelined : ?telemetry:int -> t -> extra:Neodriver_packstream.Packstream.value -> (unit, Neodriver_core.Errors.t) Stdlib.result

Send a BEGIN without reading its response (the execute_query BEGIN pipelining): the next run consumes its responses (a TELEMETRY SUCCESS, when telemetry is batched with it, and the BEGIN) before its own RUN response. The reported db is then available through take_begin_db.

val take_begin_db : t -> string option

The db the last begin_pipelined reported (Bolt 5.2+), if any; clears it.

val build_extra : ?mode:Neodriver_core.Config.access_mode -> ?db:string -> ?imp_user:string -> ?bookmarks:string list -> ?timeout:float -> ?metadata:(string * Neodriver_packstream.Packstream.value) list -> ?version:(int * int) -> ?notifications_min_severity:string -> ?notifications_disabled_categories:string list -> unit -> Neodriver_packstream.Packstream.value

The extra map for BEGIN (and auto-commit RUN): mode (Read -> "r"), db, imp_user, bookmarks, timeout (seconds, sent as tx_timeout milliseconds) and metadata (tx_metadata, already dehydrated). version (the connection's protocol version) gates the session-level notification settings (notifications_min_severity and notifications_disabled_categories) on the Bolt >= 5.2 capability; from Bolt 5.5 the categories field is named notifications_disabled_classifications.

Send a COMMIT message (end the transaction, applying its writes). Returns the full response metadata (its bookmark entry records the commit position).

val rollback : t -> (unit, Neodriver_core.Errors.t) Stdlib.result

Send a ROLLBACK message (end the transaction, discarding its writes). On a Failed connection the server already discarded the transaction implicitly, so a RESET is sent instead.

val route : ?db:string -> ?imp_user:string -> t -> routing_context:(string * string) list -> bookmarks:string list -> (Neodriver_packstream.Packstream.value, Neodriver_core.Errors.t) Stdlib.result

Fetch the routing table of db (default database when None). On Bolt 4.3+ this uses the ROUTE message; the routing_context (from the URI query) and bookmarks are sent with the request and the rt routing-table value is returned. On older servers (Bolt 3 / 4.0–4.2) the routing procedure is called instead: CALL dbms.cluster.routing.getRoutingTable($context) (Bolt 3) or CALL dbms.routing.getRoutingTable($context[, $database]) on the system database (Bolt 4.0–4.2); the single returned record is zipped with its field names into the same ttl/servers shape as the ROUTE rt value.

  • returns

    Error _ for a failed fetch (a procedure that is missing on the server is a non-fatal discovery error, e.g. a standalone 3.5 instance), or a Errors.Configuration_error for imp_user on Bolt < 4.3 (procedures do not support impersonation) and for db on Bolt 3 (no multi-db).

val pull : ?n:int -> ?qid:int -> t -> hydration:Neodriver_core.Hydration.t -> (Neodriver_core.Values.t list list * (Neodriver_packstream.Packstream.value, Neodriver_core.Errors.t) Stdlib.result, Neodriver_core.Errors.t) Stdlib.result

Send a PULL message, fetching up to n records (all by default) of the result qid (qid is omitted for the most recent query, which the server then targets). Records are hydrated with hydration. Returns the records delivered and the terminal outcome: Ok _ with the PULL summary metadata (its has_more flag, readable via Bolt.metadata_has_more, says whether more records remain) on SUCCESS, or Error _ for a server FAILURE (the records delivered before the failure are kept). A server failure leaves the connection in the Failed state.

val discard : ?n:int -> ?qid:int -> t -> ((Neodriver_packstream.Packstream.value, Neodriver_core.Errors.t) Stdlib.result, Neodriver_core.Errors.t) Stdlib.result

Send a DISCARD message, discarding up to n remaining records (all by default) of the result qid (qid is omitted for the most recent query). Returns the response: Ok (Ok _) with the DISCARD summary metadata on SUCCESS, or Error _ for a server failure (the connection enters the Failed state).

type stream

A lazily-streamed result on a connection: RUN is sent immediately, records are pulled in batches on demand. The terminal state is a summary (normal end) or an error (a server failure, surfaced after the buffered records are consumed).

val stream : ?on_complete:(Neodriver_packstream.Packstream.value -> unit) -> ?on_error:(Neodriver_core.Errors.t -> unit) -> t -> hydration:Neodriver_core.Hydration.t -> run_metadata:run_metadata -> stream

A fresh stream for the given connection, hydration scope and RUN metadata. on_complete fires with the final summary once the stream ends normally; on_error fires with the failure that terminated the stream (e.g. to mark the owning transaction as failed).

val connection : stream -> t

The connection the stream is running on.

val buffered : stream -> Neodriver_core.Values.t list list

A snapshot of the records buffered so far (still unconsumed), in order. Consuming the stream with next_record pops from the FIFO buffer.

val has_records : stream -> bool

Whether a record is buffered and available without pulling.

val next_record : stream -> Neodriver_core.Values.t list option

Pop the next buffered record, if any (None once the buffer is empty — the caller pulls for more).

val peek_record : stream -> Neodriver_core.Values.t list option

Peek at the next buffered record, if any, without consuming it.

val has_more : stream -> bool

Whether the stream still has records to pull.

val error : stream -> Neodriver_core.Errors.t option

A server failure that interrupted the stream.

The final PULL summary metadata, once the stream has ended normally.

val had_record : stream -> bool

Whether any record was pulled on this stream (used to synthesize the Bolt 4.x GQL status objects: a result with records is a Success, one without is No Data).

val run_metadata : stream -> run_metadata

The RUN metadata (field names, query id, timings, bookmark).

val pull_stream : ?n:int -> stream -> (Neodriver_core.Values.t list list, Neodriver_core.Errors.t) Stdlib.result

Pull up to n more records (all by default), buffering them, and return the newly fetched records. A server failure mid-stream is stored on the stream (error) and the records delivered before it are kept. Once the stream ends normally, its summary is stored.

  • returns

    Error _ for transport failures.

val drain_stream : stream -> unit

Pull a stream to its end, best effort: a transport failure stops the drain (the failure is left on the stream; the connection is recovered by the next request's RESET).

val discard_stream : stream -> (unit, Neodriver_core.Errors.t) Stdlib.result

Discard the rest of a stream without pulling its records (consume semantics): the DISCARD response's metadata becomes the stream's final summary. No-op once the stream is finished.

val stream_closed : stream -> bool

Whether the stream's transaction was closed (the stream is out of scope).

val mark_stream_closed : stream -> unit

Mark a stream as closed: further reads on it must fail, like the Python driver's ResultConsumedError.

val mark_stream_error : stream -> Neodriver_core.Errors.t -> unit

Mark a stream as failed with error: further reads surface the failure instead of pulling (e.g. when a sibling request failed and terminated the stream's transaction).

val server_agent : t -> string option

The server agent string reported in the HELLO response, if any.

val ssr_enabled : t -> bool

Whether the server advertised the ssr.enabled hint in its HELLO response, i.e. whether it sends rt routing tables in RUN responses (server-side routing).

val capabilities : t -> Neodriver_core.Capabilities.t

The protocol capabilities of the connection's version.

val supports_impersonation : t -> bool

Whether the connection's protocol version supports impersonation (imp_user) — Bolt 4.4 and later.

val current_auth : t -> auth option

The authentication token the connection is currently logged on with, if any.

val same_auth : t -> auth -> bool

Whether the connection is logged on with token: its current token equals token, or it carries none (it was marked unauthenticated, e.g. by an AuthorizationExpired). The re-auth paths use this to skip a redundant LOGOFF+LOGON when the token did not change.

val re_auth : ?force:bool -> t -> auth -> (bool, Neodriver_core.Errors.t) Stdlib.result

Re-authenticate when auth differs from the current token (LOGOFF then LOGON, Bolt >= 5.1). Returns whether the token changed (false when it is the same as the current one, unless force is set — which re-authenticates unconditionally, e.g. for user switching). The LOGOFF and LOGON are pipelined (auth pipelining): a subsequent request consumes their responses before its own.

  • returns

    Error _ for older protocol versions or on server failure.

val consume_pending_auth : t -> (unit, Neodriver_core.Errors.t) Stdlib.result

Read the responses of a pipelined re_auth that no request consumed yet (a no-op when none are pending); a failure drains the remaining responses and is returned. Used by callers that re-authenticated a connection without issuing a further request.

val mark_unauthenticated : t -> unit

Forget the current token (the next re_auth will log on again).