Fix session DB lock self-deadlock
ober
66eb20bbea96703e04ebc174c5b1070ad6cb7419
--- a/repl-reader.ss +++ b/repl-reader.ss @@ -1,11 +1,12 @@ #!/usr/bin/env jerboa -;;; repl-reader.ss — full-duplex talk to a jcode debug REPL over TLS. +;;; repl-reader.ss — full-duplex talk to a jcode debug REPL. ;;; ;;; Usage: ;;; ./repl-reader.ss HOST:PORT 'expr1' 'expr2' ... ;; exprs from argv ;;; ./repl-reader.ss HOST:PORT ;; exprs from stdin ;;; -;;; The auth token comes from $JCODE_REPL_TOKEN, else ./.repl-token, and is +;;; Loopback REPLs are plain TCP and do not use a token. Non-loopback REPLs +;;; use TLS and require an auth token from $JCODE_REPL_TOKEN or ./.repl-token, ;;; sent as the first line (the REPL's `token?` gate). TLS transport is ;;; `openssl s_client`, because the REPL serves a per-process self-signed ;;; cert — the token is the real auth and TLS is just wire privacy, so cert @@ -53,6 +54,38 @@ (newline (current-error-port)) (exit 2)) +(define (starts-with? s prefix) + (let ((n (string-length prefix))) + (and (>= (string-length s) n) + (string=? (substring s 0 n) prefix)))) + +(define (shell-quote s) + (let ((out (open-output-string))) + (put-char out #\') + (let lp ((i 0)) + (when (< i (string-length s)) + (let ((c (string-ref s i))) + (if (char=? c #\') + (put-string out "'\\''") + (put-char out c))) + (lp (+ i 1)))) + (put-char out #\') + (get-output-string out))) + +(define (split-target target) + (let ((n (string-length target))) + (let lp ((i 0)) + (cond + ((= i n) (die "target must be HOST:PORT")) + ((char=? (string-ref target i) #\:) + (values (substring target 0 i) + (substring target (+ i 1) n))) + (else (lp (+ i 1))))))) + +(define (loopback-host? host) + (or (string=? host "localhost") + (starts-with? host "127."))) + ;; ---- main ---- (define argv (cdr (command-line-arguments))) ;; drop the script name @@ -61,26 +94,38 @@ (define target (car argv)) (define exprs (cdr argv)) -(define token (or (read-token) - (die "no token: set JCODE_REPL_TOKEN or create .repl-token"))) + +(define-values (target-host target-port) (split-target target)) +(define plain-loopback? (loopback-host? target-host)) +(define token + (and (not plain-loopback?) + (or (read-token) + (die "no token: set JCODE_REPL_TOKEN or create .repl-token")))) (define wait (let ((w (getenv "JCODE_REPL_WAIT"))) (or (and w (string->number (trim w))) 3))) -;; `exec` so the process we spawn IS timeout (clean teardown of openssl); -;; openssl stderr (cert-verify chatter) → /dev/null; we read only stdout. +;; `exec` so the process we spawn IS timeout (clean teardown of openssl/nc); +;; transport stderr (cert-verify chatter, nc errors) -> /dev/null; read stdout. (define cmd - (string-append - "exec timeout " (number->string wait) - " openssl s_client -quiet -connect " target " 2>/dev/null")) + (if plain-loopback? + (string-append + "exec timeout " (number->string wait) + " nc " (shell-quote target-host) " " (shell-quote target-port) + " 2>/dev/null") + (string-append + "exec timeout " (number->string wait) + " openssl s_client -quiet -connect " (shell-quote target) + " 2>/dev/null"))) (call-with-values (lambda () (open-process-ports cmd (buffer-mode block) (make-transcoder (utf-8-codec)))) (lambda (to-in from-out from-err pid) - ;; send token first, then the expressions - (put-string to-in token) (put-string to-in "\n") + ;; send token first only for non-loopback TLS, then the expressions + (when token + (put-string to-in token) (put-string to-in "\n")) (if (null? exprs) (let lp () ;; exprs from our stdin, one per line (let ((line (get-line (current-input-port)))) --- a/src/jcode/core/session.ss +++ b/src/jcode/core/session.ss @@ -19,6 +19,7 @@ (import (jsqlite api) (only (jsqlite value) sql-null sql-null?) :std/misc/uuid + :std/misc/thread :std/os/path :std/text/json ./config @@ -26,6 +27,17 @@ (defstruct session (id title created messages)) +;; jsqlite uses a sidecar flock file. Concurrent session DB opens from +;; multiple jcode threads can deadlock inside flock, so serialize access +;; within this process before entering SQLite. +(def *session-db-mutex* (make-mutex)) + +(def (with-session-db-lock thunk) + (dynamic-wind + (lambda () (mutex-lock! *session-db-mutex*)) + thunk + (lambda () (mutex-unlock! *session-db-mutex*)))) + (def (db-value v) (if (sql-null? v) #f v)) @@ -51,11 +63,13 @@ ;; Run PROC on a fresh db handle, guaranteeing close even on exception. (def (with-db proc) - (let ((db (open-db))) - (dynamic-wind - (lambda () (void)) - (lambda () (proc db)) - (lambda () (sqlite-close db))))) + (with-session-db-lock + (lambda () + (let ((db (open-db))) + (dynamic-wind + (lambda () (void)) + (lambda () (proc db)) + (lambda () (sqlite-close db))))))) (def (session-init-db) (with-db @@ -173,47 +187,49 @@ (def (session-replace-messages session-id msgs) ;; Atomically replace ALL stored messages for SESSION-ID with MSGS. ;; Used by /compact to swap older turns for an LLM-generated summary. - (let* ((db (open-db)) - (now (timestamp-now)) - (begun? #f) - (committed? #f)) - (dynamic-wind - (lambda () (void)) - (lambda () - (try - (begin - (sqlite-exec db "BEGIN") - (set! begun? #t) - (sqlite-exec db "DELETE FROM messages WHERE session_id = ?" session-id) - (for-each - (lambda (msg) - (let ((tool-calls-json - (and (message-tool-calls msg) - (json-object->string - (map tool-call->stored-json (message-tool-calls msg)))))) - (sqlite-exec db - "INSERT INTO messages (session_id, role, content, tool_calls, tool_call_id, created_at) - VALUES (?, ?, ?, ?, ?, ?)" - session-id - (message-role msg) - (db-param (let ((c (message-content msg))) (if (eq? c (void)) #f c))) - (db-param tool-calls-json) - (db-param (let ((id (message-tool-call-id msg))) (if (eq? id (void)) #f id))) - now))) - msgs) - (sqlite-exec db - "UPDATE sessions SET updated_at = ? WHERE id = ?" - now session-id) - (sqlite-exec db "COMMIT") - (set! committed? #t) - (set! begun? #f)) - (catch (e) - (when (and begun? (not committed?)) - (try (sqlite-exec db "ROLLBACK") - (catch (_) (void)))) - (error 'session-replace-messages "transaction failed" e)))) - (lambda () - (sqlite-close db))))) + (with-session-db-lock + (lambda () + (let* ((db (open-db)) + (now (timestamp-now)) + (begun? #f) + (committed? #f)) + (dynamic-wind + (lambda () (void)) + (lambda () + (try + (begin + (sqlite-exec db "BEGIN") + (set! begun? #t) + (sqlite-exec db "DELETE FROM messages WHERE session_id = ?" session-id) + (for-each + (lambda (msg) + (let ((tool-calls-json + (and (message-tool-calls msg) + (json-object->string + (map tool-call->stored-json (message-tool-calls msg)))))) + (sqlite-exec db + "INSERT INTO messages (session_id, role, content, tool_calls, tool_call_id, created_at) + VALUES (?, ?, ?, ?, ?, ?)" + session-id + (message-role msg) + (db-param (let ((c (message-content msg))) (if (eq? c (void)) #f c))) + (db-param tool-calls-json) + (db-param (let ((id (message-tool-call-id msg))) (if (eq? id (void)) #f id))) + now))) + msgs) + (sqlite-exec db + "UPDATE sessions SET updated_at = ? WHERE id = ?" + now session-id) + (sqlite-exec db "COMMIT") + (set! committed? #t) + (set! begun? #f)) + (catch (e) + (when (and begun? (not committed?)) + (try (sqlite-exec db "ROLLBACK") + (catch (_) (void)))) + (error 'session-replace-messages "transaction failed" e)))) + (lambda () + (sqlite-close db))))))) (def (session-update-title session-id title) (with-db --- a/test/run.ss +++ b/test/run.ss @@ -7,6 +7,7 @@ (jcode tool task) (jcode core builtin-skills) (jcode core message) + (jcode core session) (jcode core grok-auth) (jcode core config) (jcode core escalation) @@ -49,6 +50,7 @@ (jcode eval ablation) (jcode eval runner) (std text json) + (std misc thread) (std net tcp)) ;; ── Helpers ────────────────────────────────────────────────────── @@ -72,6 +74,15 @@ (define (section name) (printf "~n~a~n" name)) +(define (with-temp-home thunk) + (let ([old-home (getenv "HOME")] + [tmp-home (format "/tmp/jcode-session-test-~a" (random 100000000))]) + (mkdir tmp-home) + (dynamic-wind + (lambda () (putenv "HOME" tmp-home)) + thunk + (lambda () (putenv "HOME" old-home))))) + ;; Build an args hashtable from flat key/value pairs (define (args . pairs) (let ([h (make-hashtable equal-hash equal?)]) @@ -255,6 +266,31 @@ (check! "tool-result content" (message-content m) "result") (check! "tool-result call-id" (message-tool-call-id m) "call-123")) +;; ── Session tests ──────────────────────────────────────────────── + +(section "=== session persistence ===") + +(with-temp-home + (lambda () + (session-init-db) + (let* ([session (session-create "concurrency regression")] + [sid (session-id session)] + [threads + (map + (lambda (i) + (spawn + (lambda () + (session-add-message sid + (make-tool-result + (format "call-~a" i) + (format "result-~a" i)))))) + '(0 1 2 3 4 5 6 7 8 9 10 11))]) + (for-each thread-join! threads) + (check! "concurrent session writes persist all messages" + (length (session-get-messages sid)) + 12) + (session-delete sid)))) + ;; ── File tool tests ─────────────────────────────────────────────── (section "=== file tools ===")