provider: time out streaming reads after 90s of silence
ober
b9fe946b64fbba14fc759f7eee33d680cfcb1368
--- a/src/jcode/provider/provider.ss +++ b/src/jcode/provider/provider.ss @@ -300,21 +300,58 @@ (close-port in) (close-port out)))))))) +;; Streaming read timeout: if no bytes arrive within this many seconds a +;; watchdog thread closes the connection so the read loop unblocks and we +;; raise a clear error instead of hanging indefinitely. Override at runtime +;; via (parameterize ((*stream-read-timeout-secs* N)) ...). +(def *stream-read-timeout-secs* (make-parameter 90)) + ;; Streaming HTTP POST: calls line-cb with each line of the response body. ;; Used for SSE (Server-Sent Events) streaming from LLM APIs. (def (jcode-http-post-stream url headers body-json line-cb) (let-values (((scheme host port path) (parse-url-parts url))) - (let ((req (build-http-request "POST" path host headers body-json))) + (let ((req (build-http-request "POST" path host headers body-json)) + (timeout-secs (*stream-read-timeout-secs*))) (if (equal? scheme "https") - ;; HTTPS via rustls - (let ((conn (rustls-connect host port))) + ;; HTTPS via rustls -- guarded by a watchdog thread that closes the + ;; connection if no bytes arrive for timeout-secs seconds. + (let ((conn (rustls-connect host port)) + (last-activity (vector (time-second (current-time)))) + (timed-out? (vector #f)) + (done? (vector #f)) + (closed? (vector #f))) + (def (touch!) + (vector-set! last-activity 0 (time-second (current-time)))) + (def (close-once!) + (unless (vector-ref closed? 0) + (vector-set! closed? 0 #t) + (rustls-close conn))) + (fork-thread + (lambda () + (let loop () + (sleep (make-time 'time-duration 0 5)) + (cond + ((vector-ref done? 0) (void)) + ((>= (- (time-second (current-time)) + (vector-ref last-activity 0)) + timeout-secs) + (vector-set! timed-out? 0 #t) + (when (tracing?) + (log-trace logger "stream-timeout" + `((url . ,(redact-url url)) + (idle-secs . ,(- (time-second (current-time)) + (vector-ref last-activity 0)))))) + (close-once!)) + (else (loop)))))) (dynamic-wind (lambda () (void)) (lambda () (tls-write-string conn req) + (touch!) (let* ((status-line (tls-read-line conn)) (status (parse-http-status status-line)) (_headers (read-tls-headers conn))) + (touch!) (unless (= status 200) (let ((body (tls-read-all conn))) (error 'jcode-http-post-stream @@ -322,12 +359,19 @@ ;; Read SSE lines until EOF or chunked terminator (let loop () (let ((line (tls-read-line conn))) + (touch!) (when line (unless (equal? line "0") ;; chunked transfer end (line-cb line) (loop))))) + (when (vector-ref timed-out? 0) + (error 'jcode-http-post-stream + (format "stream read timed out after ~as of silence (host: ~a)" + timeout-secs host))) status)) - (lambda () (rustls-close conn)))) + (lambda () + (vector-set! done? 0 #t) + (close-once!)))) ;; Plain HTTP via tcp (let-values (((in out) (tcp-connect host port))) (dynamic-wind