resolved conflict

private-player-agent/http-transfer.rkt

@@ -0,0 +1,120 @@
+#lang racket/base
+
+(require racket/async-channel)
+
+(provide call-with-http-timeout)
+
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+; goal   : Bound HTTP inactivity while allowing transfers of any duration.
+; pre    : Seconds is positive. Proc consumes its HTTP ports before returning
+;          and calls progress! whenever new response bytes arrive.
+; post   : The worker has exited and its HTTP resources have been closed.
+; result : Proc's values, or its exception, or an inactivity timeout error.
+; internals: Activity restarts sync/timeout. On cancellation, a break unwinds
+;            semaphore and dynamic-wind cleanup before the custodian closes
+;            remaining ports. No timed-out worker can continue changing state.
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+(define (call-with-http-timeout seconds proc)
+  (let* ((custodian (make-custodian))
+         (activity (make-async-channel 1))
+         (result (make-async-channel))
+         (finished? #f)
+         (worker
+          (parameterize ((current-custodian custodian))
+            (thread
+             (λ ()
+               (with-handlers ((exn:break? void))
+                 (parameterize-break #t
+                   (with-handlers
+                       ((exn? (λ (exception)
+                                (async-channel-put result (cons #f exception)))))
+                     (let ((values-list
+                            (call-with-values
+                             (λ () (proc (λ () (async-channel-put activity #t))))
+                             list)))
+                       (async-channel-put result (cons #t values-list)))))))))))
+    (dynamic-wind
+      void
+      (λ ()
+        (let loop ()
+          (let ((event (sync/timeout seconds
+                                    (handle-evt result (λ (value) (cons 'result value)))
+                                    (handle-evt activity (λ (_) 'progress)))))
+            (cond
+              ((not event)
+               (error 'http-transfer "No HTTP progress for ~a seconds" seconds))
+              ((eq? event 'progress) (loop))
+              (else
+               (set! finished? #t)
+               (if (cadr event)
+                   (apply values (cddr event))
+                   (raise (cddr event))))))))
+      (λ ()
+        (unless (or finished? (thread-dead? worker))
+          (break-thread worker))
+        (thread-wait worker)
+        (custodian-shutdown-all custodian)))))
+
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+;; Tests for module http-transfer.rkt
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+
+(module+ test
+  (require rackunit racket/file)
+
+  (check-equal? (call-with-values
+                (λ () (call-with-http-timeout 1 (λ (_) (values 1 2)))) list)
+                '(1 2))
+  (check-exn #rx"original failure"
+             (λ () (call-with-http-timeout 1 (λ (_) (error "original failure")))))
+
+  (test-case "continuing progress allows a transfer longer than its idle limit"
+    (check-equal?
+     (call-with-http-timeout
+      0.5
+      (λ (progress!)
+        (let loop ((chunks 6))
+          (unless (zero? chunks)
+            (sleep 0.15)
+            (progress!)
+            (loop (- chunks 1))))
+        'downloaded))
+     'downloaded))
+
+  (test-case "stalled HTTP workers release locks and close their ports"
+    (let ((lock (make-semaphore 1))
+          (output #f)
+          (target (make-temporary-file "http-stall-~a")))
+      (dynamic-wind
+        void
+        (λ ()
+          (check-exn
+           #rx"No HTTP progress"
+           (λ ()
+             (parameterize-break #f
+               (call-with-http-timeout
+                0.1
+                (λ (_)
+                  (set! output (open-output-file target #:exists 'truncate))
+                  (call-with-semaphore lock (λ () (sync never-evt))))))))
+          (check-true (semaphore-try-wait? lock))
+          (check-true (port-closed? output)))
+        (λ () (delete-file target)))))
+
+  (test-case "breaking the caller also unwinds its HTTP worker"
+    (let* ((ready (make-semaphore))
+           (lock (make-semaphore 1))
+           (caller
+            (thread
+             (λ ()
+               (with-handlers ((exn:break? void))
+                 (call-with-http-timeout
+                  5
+                  (λ (_)
+                    (call-with-semaphore
+                     lock
+                     (λ () (semaphore-post ready) (sync never-evt))))))))))
+      (semaphore-wait ready)
+      (break-thread caller)
+      (check-not-false (sync/timeout 2 caller))
+      (check-true (semaphore-try-wait? lock)))))

private-player-agent/player-agent-core.rkt

@@ -9,7 +9,8 @@
          racket/port
          racket/string
          simple-log
-         "player-agent-translate.rkt")
+         "player-agent-translate.rkt"
+         "http-transfer.rkt")
 
 (provide (struct-out player-agent-runtime)
          make-player-agent-runtime)
@@ -44,6 +45,9 @@
 ;;; Posts JSON to an agent API endpoint and reads its JSON response.
 ;;; Authorization failures receive a distinct exception for poll-loop.
 (define (post-json base path data)
+  (call-with-http-timeout
+   30
+   (λ (_progress!)
      (let ((input
             (post-pure-port
              (endpoint-url base path)
@@ -64,7 +68,7 @@
                      (current-continuation-marks)))
                    (error 'player-agent (hash-ref response 'error))))
              response))
-      (λ () (close-input-port input)))))
+         (λ () (close-input-port input)))))))
 
 ;;; Converts racket-audio states to the state names sent to the web player.
 (define (normal-state state)
@@ -271,7 +275,7 @@
                             (bytes->string/utf-8 extension))))
            (path (format "/api/agent/media/~a/~a" app-id token))
            (input #f))
-      (with-handlers ((exn:fail?
+      (with-handlers ((exn?
                        (λ (exception)
                          (when input (close-input-port input))
                          (safe-delete-file target)
@@ -285,6 +289,9 @@
                  (hasheq 'filename filename 'received 0
                          'total (if (exact-nonnegative-integer? size)
                                     size 'null)))))
+        (call-with-http-timeout
+         30
+         (λ (progress!)
           (set! input (get-pure-port (endpoint-url server-url path)))
           (call-with-output-file
            target
@@ -292,12 +299,13 @@
              (copy-media-port
               input output
               (λ (received)
+                (progress!)
                 (with-agent-state
                  (λ ()
                    (set! download-progress
                          (hash-set download-progress 'received received)))))))
            #:exists 'truncate/replace)
-        (close-input-port input)
+          (close-input-port input)))
         (when (and (exact-nonnegative-integer? size)
                    (not (= (file-size target) size)))
           (error 'player-agent
@@ -558,6 +566,8 @@
               (set! command-worker
                     (thread
                      (λ ()
+                       (with-handlers ((exn:break? void))
+                         (parameterize-break #t
                            (with-handlers
                                ((exn:fail?
                                  (λ (exception)
@@ -588,7 +598,7 @@
                              (execute-command! command))
                            (set! acknowledged-command (hash-ref command 'id))
                            (set! executing-command-id 0)
-                       (set! command-worker #f))))))
+                           (set! command-worker #f))))))))
           (sleep 1)
           (loop)))))
 
@@ -597,15 +607,23 @@
     (unless running
       (set! running #t)
       (status-callback (tr 'connecting))
-      (set! worker (thread poll-loop))))
+      (set! worker
+            (thread
+             (λ ()
+               (with-handlers ((exn:break? void))
+                 (parameterize-break #t (poll-loop))))))))
 
   ;;; Stops polling and command workers and clears their lifecycle state.
   (define (stop!)
     (set! running #f)
-    (when (and worker (not (thread-dead? worker)))
-      (kill-thread worker))
-    (when (and command-worker (not (thread-dead? command-worker)))
-      (kill-thread command-worker))
+    (let ((poll-worker worker)
+          (active-command command-worker))
+      (when (and poll-worker (not (thread-dead? poll-worker)))
+        (break-thread poll-worker)
+        (thread-wait poll-worker))
+      (when (and active-command (not (thread-dead? active-command)))
+        (break-thread active-command)
+        (thread-wait active-command)))
     (set! worker #f)
     (set! command-worker #f)
     (with-agent-state (λ () (set! download-progress 'null)))

private/dlna-playback.rkt

@@ -359,7 +359,10 @@
                          (make-semaphore 1))))
     (set-dlna-playback-monitor!
      playback
-     (thread (λ () (monitor-loop playback poll-seconds))))
+     (thread (λ ()
+               (with-handlers ((exn:break? void))
+                 (parameterize-break #t
+                   (monitor-loop playback poll-seconds))))))
     playback))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
@@ -483,7 +486,9 @@
     (set-dlna-playback-running?! playback #f)
     (let ((monitor (dlna-playback-monitor playback)))
       (when (and monitor (not (thread-dead? monitor)))
-        (kill-thread monitor))
+        ;; A break unwinds call-with-semaphore; kill-thread leaves its lock held.
+        (break-thread monitor)
+        (thread-wait monitor))
       (set-dlna-playback-monitor! playback #f)
       (with-lock
        playback

private/player.rkt

@@ -1461,6 +1461,11 @@
                         (player-renderers value))
         'rendererId (or (playback-session-selected-id session) 'null)
         'currentIndex (or (playback-session-current-index session) 'null)
+        'download
+        (let ((backend (playback-session-backend session)))
+          (if (and backend (eq? (playback-session-backend-kind session) 'agent))
+              (hash-ref (playback-agent-reported-state backend) 'download 'null)
+              'null))
         'state (symbol->string (playback-session-state session))
         'position (playback-session-position session)
         'duration (or (playback-session-duration session) 'null)
@@ -2154,7 +2159,13 @@
                    'state
                    (hasheq 'state "stopped"
                            'position 0
-                           'volume 25))))
+                           'volume 25
+                           'download (hasheq 'filename "Large.flac"
+                                             'received 65536 'total 100000000)))))
+
+        (check-equal?
+         (hash-ref (player-state->jsexpr example-player) 'download)
+         (hasheq 'filename "Large.flac" 'received 65536 'total 100000000))
 
         (check-equal?
          (hash-ref (hash-ref first-poll 'command) 'action)

private/server.rkt

@@ -7,6 +7,8 @@
          racket/runtime-path
          racket-mimetypes
          racket/string
+         racket/path
+         simple-log
          net/url
          web-server/dispatch
          web-server/http
@@ -19,6 +21,8 @@
 
 (provide serve-player)
 
+(sl-def-log web-player-http)
+
 (define-runtime-path public-directory "../public")
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
@@ -176,16 +180,47 @@
       player
       (request-jsexpr request)))))
 
+;;; Streams an agent download without holding player locks and reports its
+;;; start, progress and outcome, including the client and actual bytes sent.
+(define (stream-agent-media! file output client app-id)
+  (let ((received 0)
+        (total (file-size file))
+        (started (current-inexact-monotonic-milliseconds))
+        (reported (current-inexact-monotonic-milliseconds))
+        (completed? #f)
+        (name (file-name-from-path file)))
+    (dynamic-wind
+      (λ () (info-web-player-http "Agent download started: client=~a agent=~a file=~a bytes=~a"
+                                  client (substring app-id 0 8) name total))
+      (λ ()
+        (call-with-input-file
+         file
+         (λ (input)
+           (let ((buffer (make-bytes 65536)))
+             (let loop ()
+               (let ((count (read-bytes-avail! buffer input)))
+                 (unless (eof-object? count)
+                   (write-bytes buffer output 0 count)
+                   (set! received (+ received count))
+                   (when (>= (- (current-inexact-monotonic-milliseconds) reported) 5000)
+                     (set! reported (current-inexact-monotonic-milliseconds))
+                     (info-web-player-http "Agent download progress: client=~a file=~a bytes=~a/~a"
+                                           client name received total))
+                   (loop)))))))
+        (flush-output output)
+        (set! completed? #t))
+      (λ ()
+        (info-web-player-http "Agent download ~a: client=~a file=~a bytes=~a/~a elapsed-ms=~a"
+                              (if completed? "sent" "interrupted") client name received total
+                              (round (- (current-inexact-monotonic-milliseconds) started)))))))
+
 ;;; Streams the media file identified by an agent's opaque token.
-(define (agent-media-handler player _request app-id token)
+(define (agent-media-handler player request app-id token)
   (let ((file (player-agent-media player app-id token)))
     (if (and file (file-exists? file))
         (response/output
          (λ (output)
-           (call-with-input-file
-            file
-            (λ (input)
-              (copy-port input output))))
+           (stream-agent-media! file output (request-client-ip request) app-id))
          #:mime-type
          (let ((mime (mimetype-for-ext file)))
            (if (string? mime)

public/app.js

@@ -146,12 +146,21 @@ async function api(path, body) {
     };
   }
 
-  const response = await fetch(path, options);
+  const controller = new AbortController();
+  const timeout = setTimeout(() => controller.abort(), 30000);
+  try {
+    const response = await fetch(path, { ...options, signal: controller.signal });
     const data = await response.json();
     if (!response.ok) {
       throw new ApiError(data.error || `HTTP ${response.status}`, response.status, data.code);
     }
     return data;
+  } catch (error) {
+    if (controller.signal.aborted) throw new Error("request-timeout");
+    throw error;
+  } finally {
+    clearTimeout(timeout);
+  }
 }
 
 // Displays the login overlay and optionally reports why authentication failed.
@@ -671,6 +680,11 @@ function render(nextState) {
   renderPlayer(nextState);
   if (nextState.error) {
     setStatus(t(nextState.error));
+  } else if (nextState.download) {
+    const download = nextState.download;
+    const received = (download.received / 1000000).toFixed(1);
+    const total = Number.isFinite(download.total) ? (download.total / 1000000).toFixed(1) : "?";
+    setStatus(t("downloadingMedia", { name: download.filename, received, total }));
   } else {
     setStatus(nextState.discovering ? t("searchingPlayers") : "");
   }

public/translate.js

@@ -3,6 +3,8 @@
 
   const translations = {
     en: {
+      "request-timeout": "The server did not respond within 30 seconds. Status will be refreshed automatically.",
+      downloadingMedia: "Downloading {name}: {received} / {total} MB",
       folders: "Folders",
       savePlaylist: "SAVE PLAYLIST",
       playlistSaved: "SAVED",
@@ -41,6 +43,8 @@
       "dlna-renderer-unreachable": "The DLNA renderer is unreachable.",
     },
     nl: {
+      "request-timeout": "De server antwoordde niet binnen 30 seconden. De status wordt automatisch opnieuw opgehaald.",
+      downloadingMedia: "Downloaden van {name}: {received} / {total} MB",
       folders: "Mappen",
       savePlaylist: "AFSPEELLIJST OPSLAAN",
       playlistSaved: "OPGESLAGEN",

scribblings/rkt-web-player.scrbl

@@ -143,6 +143,23 @@ the agent exits; it is not retained across restarts.
 A track that is not cached is still downloaded completely before playback.
 The HTTP server allows up to one hour for a response, including media
 transfers, so large lossless tracks can finish over slower connections.
+The agent imposes no total download deadline: receiving another chunk restarts
+its 30-second inactivity timer. A connection or transfer with no response bytes
+for 30 seconds fails, closes its ports and removes the partial file. The private
+@tt{call-with-http-timeout} procedure runs an HTTP operation with a
+@tt{progress!} callback; each callback restarts this timer. Cancellation unwinds
+the worker before its custodian closes the remaining HTTP resources.
+
+The browser displays the selected agent's reported download filename and byte
+counts, including prefetch downloads. Server logs record agent download start,
+progress every five seconds while sending, and completion or interruption with
+client address, filename and byte counts. DLNA media transfers are logged by
+@tt{racket-upnp} with client address, method, range and elapsed time.
+
+Browser API requests time out after 30 seconds, freeing the request queue and
+allowing status polling to resume. Timing out a browser request does not undo a
+command already accepted by the server.
+
 The next track is fetched after the play command completes. Commands are
 processed in server order, so a pending prefetch can still delay a later manual
 track selection.