Keep non-stream provider waits observable
ober
bcb0213de0c5b8c16d4226b6322931f468806e73
--- a/src/jcode/provider/provider.ss +++ b/src/jcode/provider/provider.ss @@ -490,14 +490,46 @@ ;; Generic HTTP POST with JSON body, returns (values status body-string) (def (http-post-json url headers body-json) (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 (http-read-timeout-for-host host))) (if (equal? scheme "https") - ;; HTTPS via rustls - (let ((conn (rustls-connect host port))) + ;; HTTPS via rustls -- guarded against silent non-streaming reads. + (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) + (jcode-tls-close conn))) + (fork-thread + (lambda () + (guard (e [#t (when (tracing?) + (log-trace logger "http-watchdog-died" + `((err . ,(err->string e)))))]) + (let loop () + (thread-sleep! 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 "http-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)) (resp-headers (read-tls-headers conn)) @@ -510,14 +542,52 @@ (cl (tls-read-n conn (string->number (cdr cl)))) (chunked? (tls-read-chunked conn)) (else (tls-read-all conn))))) + (when (vector-ref timed-out? 0) + (error 'http-post-json + (format "HTTP read timed out after ~as of silence (host: ~a)" + timeout-secs host))) (values status body))) - (lambda () (jcode-tls-close conn)))) + (lambda () + (vector-set! done? 0 #t) + (close-once!)))) ;; Plain HTTP via tcp (let-values (((in out) (tcp-connect host port))) + (let ((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) + (guard (e [(i/o-error? e) (void)]) (close-port in)) + (guard (e [(i/o-error? e) (void)]) (close-port out)))) + (fork-thread + (lambda () + (guard (e [#t (when (tracing?) + (log-trace logger "http-watchdog-died" + `((err . ,(err->string e)))))]) + (let loop () + (thread-sleep! 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 "http-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 () (port-write-string out req) + (touch!) ;; Honor Content-Length / chunked like the TLS branch above. ;; A bare read-all hangs forever on keep-alive servers (e.g. ;; ollama answers and holds the socket open — no EOF arrives). @@ -533,10 +603,14 @@ (cl (port-read-n in (string->number (cdr cl)))) (chunked? (port-read-chunked in)) (else (port-read-all in))))) + (when (vector-ref timed-out? 0) + (error 'http-post-json + (format "HTTP read timed out after ~as of silence (host: ~a)" + timeout-secs host))) (values status body))) (lambda () - (close-port in) - (close-port out)))))))) + (vector-set! done? 0 #t) + (close-once!))))))))) ;; Streaming read timeout: if no bytes arrive within this many seconds a ;; watchdog thread closes the connection so the read loop unblocks and we @@ -548,6 +622,8 @@ ;; via (parameterize ((*stream-read-timeout-secs* N)) ...). (def *stream-read-timeout-secs* (make-parameter 45)) (def *local-stream-read-timeout-floor-secs* 300) +(def *http-read-timeout-secs* (make-parameter 600)) +(def *local-http-read-timeout-floor-secs* 600) (def (loopback-host? host) (or (equal? host "127.0.0.1") @@ -560,37 +636,44 @@ (max base *local-stream-read-timeout-floor-secs*) base))) +(def (http-read-timeout-for-host host) + (let ((base (*http-read-timeout-secs*))) + (if (loopback-host? host) + (max base *local-http-read-timeout-floor-secs*) + base))) + (def *provider-wait-heartbeat-secs* 30) -(def (call-with-local-provider-heartbeat provider url thunk) +(def (call-with-provider-heartbeat provider url thunk) (let-values (((_scheme host _port _path) (parse-url-parts url))) - (if (loopback-host? host) - (let ((done? (vector #f)) - (err-port (current-error-port)) - (log-lvl (current-log-level)) - (started (time-second (current-time)))) - (fork-thread - (lambda () - (parameterize ((current-error-port err-port) - (current-log-level log-lvl)) - (guard (e [#t (when (tracing?) - (log-trace logger "provider-heartbeat-died" - `((err . ,(err->string e)))))]) - (let loop () - (thread-sleep! *provider-wait-heartbeat-secs*) - (unless (vector-ref done? 0) - (log-info logger "waiting-for-local-provider" - `((provider . ,(provider-name provider)) - (model . ,(provider-model provider)) - (elapsed . ,(format "~as" - (- (time-second (current-time)) - started))))) - (loop))))))) - (dynamic-wind - (lambda () (void)) - thunk - (lambda () (vector-set! done? 0 #t)))) - (thunk)))) + (let ((done? (vector #f)) + (err-port (current-error-port)) + (log-lvl (current-log-level)) + (started (time-second (current-time)))) + (fork-thread + (lambda () + (parameterize ((current-error-port err-port) + (current-log-level log-lvl)) + (guard (e [#t (when (tracing?) + (log-trace logger "provider-heartbeat-died" + `((err . ,(err->string e)))))]) + (let loop () + (thread-sleep! *provider-wait-heartbeat-secs*) + (unless (vector-ref done? 0) + (log-info logger + (if (loopback-host? host) + "waiting-for-local-provider" + "waiting-for-provider") + `((provider . ,(provider-name provider)) + (model . ,(provider-model provider)) + (elapsed . ,(format "~as" + (- (time-second (current-time)) + started))))) + (loop))))))) + (dynamic-wind + (lambda () (void)) + thunk + (lambda () (vector-set! done? 0 #t)))))) ;; Streaming HTTP POST: calls line-cb with each line of the response body. ;; Used for SSE (Server-Sent Events) streaming from LLM APIs. @@ -1042,7 +1125,7 @@ (headers . ,(redact-headers headers)) (body . ,body-json)))) (let-values (((status text) - (call-with-local-provider-heartbeat provider url + (call-with-provider-heartbeat provider url (lambda () (http-post-json url headers body-json))))) (when (tracing?) (log-trace logger "openai-response" @@ -1158,7 +1241,7 @@ (headers . ,(redact-headers headers)) (body . ,body-json)))) (let-values (((status text) - (call-with-local-provider-heartbeat provider url + (call-with-provider-heartbeat provider url (lambda () (http-post-json url headers body-json))))) (when (tracing?) (log-trace logger "openai-response"