distribution: download-update progress reporting

CHANGELOG.md

@@ -2,6 +2,14 @@
 
 ## Unreleased
 
+- `download-update` reports download progress: an optional `#:on-progress`
+  callback receives cumulative bytes from zero through the final downloaded
+  size. Intermediate reports are throttled to about one per percent of the
+  signed artifact size by default, with an optional byte cadence for activity
+  watchdogs. In-app updaters can drive progress bars without copying Rivet's
+  bounded, redirect-aware downloader. Interrupted or cancelled downloads
+  remove their partial file as reliably as ordinary failures.
+
 - Add a Windows `rivet::system::TrayController` that owns tray window
   subclassing, localized native menu presentation, command dispatch, and
   automatic `TaskbarCreated` recovery. The reference Taskboard now exercises

rivet/distribution/updater.rkt

@@ -35,18 +35,41 @@
   '(prepared installing installed restarting restarted checking healthy
              committing committed rollback-started rolled-back rollback-failed))
 
-(define (copy-limited! in out limit)
+(define (copy-limited! in out limit
+                       #:who [who 'copy-limited!]
+                       #:on-progress [on-progress void]
+                       #:progress-every [progress-every 1])
+  ;; Report the initial zero and the final byte count. Intermediate reports are
+  ;; separated by at least progress-every bytes so fast links do not flood UI
+  ;; pollers. This helper stays private; download-update is the public contract.
   (define buffer (make-bytes 65536))
-  (let loop ([total 0])
+  (on-progress 0)
+  (let loop ([total 0] [last-reported 0])
     (define count (read-bytes-avail! buffer in))
     (cond
-      [(eof-object? count) total]
+      [(eof-object? count)
+       (unless (= last-reported total) (on-progress total))
+       total]
       [else
        (define next (+ total count))
        (when (> next limit)
-         (error 'download-update "update exceeds configured download limit"))
+         (error who "response exceeds permitted byte limit"))
        (write-bytes buffer out 0 count)
-       (loop next)])))
+       (when (>= (- next last-reported) progress-every)
+         (on-progress next)
+         (set! last-reported next))
+       (loop next last-reported)])))
+
+(define (call-with-partial-cleanup temporary thunk)
+  (with-handlers ([exn?
+                   (lambda (error)
+                     (when (file-exists? temporary) (delete-file temporary))
+                     (raise error))])
+    (thunk)))
+
+(module+ test
+  (provide copy-limited!
+           call-with-partial-cleanup))
 
 ;; GitHub release assets — the dominant update origin — answer with a 302
 ;; to their CDN, so every fetch must follow redirections or the updater
@@ -66,7 +89,7 @@
     void
     (lambda ()
       (define out (open-output-bytes))
-      (copy-limited! in out maximum-bytes)
+      (copy-limited! in out maximum-bytes #:who 'fetch-update-manifest)
       (verify-signed-manifest (open-input-bytes (get-output-bytes out))
                               public-key
                               #:key-id key-id))
@@ -125,33 +148,56 @@
                            "actual" actual))
   path)
 
-(define (download-update config candidate destination)
+(define (download-update config candidate destination
+                         #:on-progress [on-progress void]
+                         #:progress-every [configured-progress-every #f])
+  (unless (and (procedure? on-progress)
+               (procedure-arity-includes? on-progress 1))
+    (raise-argument-error 'download-update
+                          "procedure accepting one argument"
+                          on-progress))
+  (unless (or (not configured-progress-every)
+              (exact-positive-integer? configured-progress-every))
+    (raise-argument-error 'download-update
+                          "(or/c #f exact-positive-integer?)"
+                          configured-progress-every))
   (define artifact (update-candidate-artifact candidate))
   (define maximum (updater-config-maximum-download-bytes config))
   (when (> (update-artifact-size artifact) maximum)
     (error 'download-update "signed artifact size exceeds configured download limit"))
   (make-parent-directory* destination)
   (define temporary (path-add-extension destination #".partial"))
   (when (file-exists? temporary) (delete-file temporary))
-  (with-handlers ([exn:fail?
-                   (lambda (e)
-                     (when (file-exists? temporary) (delete-file temporary))
-                     (raise e))])
-    (define in
-      (get-pure-port (string->url (update-artifact-url artifact))
-                     '("User-Agent: Rivet-Updater/1")
-                     #:redirections update-fetch-redirections))
-    (dynamic-wind
-      void
-      (lambda ()
-        (call-with-output-file temporary
-          #:exists 'truncate/replace
-          #:mode 'binary
-          (lambda (out) (copy-limited! in out maximum))))
-      (lambda () (close-input-port in)))
-    (verify-update-artifact! candidate temporary)
-    (rename-file-or-directory temporary destination #t)
-    destination))
+  (call-with-partial-cleanup
+   temporary
+   (lambda ()
+     (define in
+       (get-pure-port (string->url (update-artifact-url artifact))
+                      '("User-Agent: Rivet-Updater/1")
+                      #:redirections update-fetch-redirections))
+     (dynamic-wind
+       void
+       (lambda ()
+         (call-with-output-file temporary
+           #:exists 'truncate/replace
+           #:mode 'binary
+           (lambda (out)
+             ;; Bound the stream to the signed size, not merely the looser
+             ;; configured ceiling, so progress never exceeds 100% before a
+             ;; malformed response is rejected.
+             (copy-limited! in out (update-artifact-size artifact)
+                            #:who 'download-update
+                            #:on-progress on-progress
+                            #:progress-every
+                            (or configured-progress-every
+                                (max 1
+                                     (quotient
+                                      (update-artifact-size artifact)
+                                      100)))))))
+       (lambda () (close-input-port in)))
+     (verify-update-artifact! candidate temporary)
+     (rename-file-or-directory temporary destination #t)
+     destination)))
 
 (define (make-install-plan candidate downloaded-path
                            #:backup-path [backup-path #f]

rivet/scribblings/rivet.scrbl

@@ -441,10 +441,21 @@ Applies identity, channel, version, minimum-version, rollout, platform, and
 architecture policy.}
 @defproc[(download-update [config updater-config?]
                           [candidate update-candidate?]
-                          [destination path-string?])
+                          [destination path-string?]
+                          [#:on-progress on-progress procedure? void]
+                          [#:progress-every progress-every
+                           (or/c #f exact-positive-integer?) #f])
          path?]{
 Downloads to a partial file, enforces the signed size and configured bound,
-verifies SHA-256, and atomically moves the verified artifact into place.}
+verifies SHA-256, and atomically moves the verified artifact into place.
+@racket[on-progress] receives cumulative downloaded bytes, starting at zero
+and ending at the downloaded size. Intermediate reports are throttled to about
+one per percent of the signed artifact size. The callback runs synchronously
+on the downloading thread and should quickly dispatch or record UI work. A
+final byte count reports transfer progress, not successful verification.
+Pass @racket[progress-every] to select a byte cadence; @racket[#f] uses the
+one-percent default. Small cadences are appropriate for an activity watchdog,
+not direct UI rendering.}
 @defproc[(execute-install-plan! [plan install-plan?]
                                 [#:health-check health-check (-> any/c) (lambda () #t)]
                                 [#:commit commit (-> any/c) void]

tests/distribution.rkt

@@ -6,7 +6,11 @@
          rackunit
          racket/file
          racket/port
-         "../rivet/distribution.rkt")
+         "../rivet/distribution.rkt"
+         (rename-in (submod "../rivet/distribution/updater.rkt" test)
+                    [copy-limited! copy-limited!/test]
+                    [call-with-partial-cleanup
+                     call-with-partial-cleanup/test]))
 
 (use-all-factories!)
 
@@ -64,6 +68,19 @@
              (select-update linux-config linux-manifest
                             #:install-kind 'unknown)))
 
+(check-exn exn:fail:contract?
+           (lambda ()
+             (download-update linux-config
+                              (update-candidate linux-manifest linux-zip)
+                              "unused"
+                              #:on-progress (lambda () (void)))))
+(check-exn exn:fail:contract?
+           (lambda ()
+             (download-update linux-config
+                              (update-candidate linux-manifest linux-zip)
+                              "unused"
+                              #:progress-every 0)))
+
 (when private-key
   (define public-key
     (datum->pk-key (pk-key->datum private-key 'SubjectPublicKeyInfo)
@@ -316,3 +333,44 @@
     (check-equal? commits 2)
     (check-false (file-exists? journal-path)))
   (lambda () (delete-directory/files commit-root)))
+
+;; download-update's private copy path streams the payload, enforces the byte
+;; limit, and reports a useful zero-to-final sequence without flooding the
+;; callback. A payload larger than the copy buffer makes the cadence assertion
+;; exercise multiple reads instead of passing vacuously on one read.
+(let* ([payload (apply bytes (for/list ([i (in-range 200000)]) (modulo i 256)))]
+       [in (open-input-bytes payload)]
+       [out (open-output-bytes)]
+       [reports '()])
+  (define total
+    (copy-limited!/test in out (* 1024 1024)
+                        #:on-progress
+                        (lambda (n) (set! reports (cons n reports)))
+                        #:progress-every 100000))
+  (check-equal? total 200000)
+  (check-equal? (get-output-bytes out) payload)
+  (check-equal? (reverse reports) '(0 131072 200000)))
+
+(let ([in (open-input-bytes (make-bytes 10))]
+      [out (open-output-bytes)])
+  ;; The limit is enforced before the offending chunk is written.
+  (check-exn #rx"permitted byte limit"
+             (lambda ()
+               (copy-limited!/test in out 5)))
+  (check-equal? (bytes-length (get-output-bytes out)) 0))
+
+;; Cancellation uses exn:break rather than exn:fail. It must still remove the
+;; partial download before the worker publishes its terminal state.
+(define interrupted-partial
+  (make-temporary-file "rivet-interrupted-download-~a.partial"))
+(check-exn
+ exn:break?
+ (lambda ()
+   (call-with-partial-cleanup/test
+    interrupted-partial
+    (lambda ()
+      (let/ec escape
+        (raise (exn:break "cancelled"
+                          (current-continuation-marks)
+                          escape)))))))
+(check-false (file-exists? interrupted-partial))