offload: '{send,receive}-files' wait for completion of the transfer.

Fixes situations where the remote 'guix build' is invoked before the
.drv has been completely copied, as reported at
<https://lists.gnu.org/archive/html/guix-devel/2014-04/msg00174.html>.

In some cases 'send-files' would return before the other end is done
importing the files, and so the subsequent 'guix build' invocation would
just miss the .drv file it refers to.

* guix/utils.scm (call-with-decompressed-port): Don't close PORT.
  (call-with-compressed-output-port): Likewise.
* tests/utils.scm ("compressed-output-port + decompressed-port"): Adjust
  accordingly.
* guix/scripts/offload.scm (send-files): Add explicit (close-pipe pipe)
  call.
  (retrieve-files): Likewise.

guix/scripts/offload.scm

@@ -474,7 +474,9 @@ success, #f otherwise."
                 (warning (_ "failed while exporting files to '~a': ~a~%")
                          (build-machine-name machine)
                          (strerror (system-error-errno args)))))))
-        #t))))
+
+        ;; Wait for the 'lsh' process to complete.
+        (zero? (close-pipe pipe))))))
 
 (define (retrieve-files files machine)
   "Retrieve FILES from MACHINE's store, and import them."
@@ -502,7 +504,8 @@ success, #f otherwise."
                                    #:log-port (current-error-port)
                                    #:lock? #f)))
 
-             #t)))))
+             ;; Wait for the 'lsh' process to complete.
+             (zero? (close-pipe pipe)))))))
 
 
 ;;;

guix/utils.scm

@@ -229,14 +229,12 @@ a symbol such as 'xz."
 
 (define (call-with-decompressed-port compression port proc)
   "Call PROC with a wrapper around PORT, a file port, that decompresses data
-read from PORT according to COMPRESSION, a symbol such as 'xz.  PORT is closed
+read from PORT according to COMPRESSION, a symbol such as 'xz."
-as soon as PROC's dynamic extent is entered."
   (let-values (((decompressed pids)
                 (decompressed-port compression port)))
     (dynamic-wind
       (const #f)
       (lambda ()
-        (close-port port)
         (proc decompressed))
       (lambda ()
         (close-port decompressed)
@@ -286,14 +284,12 @@ of PIDs to wait for."
 
 (define (call-with-compressed-output-port compression port proc)
   "Call PROC with a wrapper around PORT, a file port, that compresses data
-that goes to PORT according to COMPRESSION, a symbol such as 'xz.  PORT is
+that goes to PORT according to COMPRESSION, a symbol such as 'xz."
-closed as soon as PROC's dynamic extent is entered."
   (let-values (((compressed pids)
                 (compressed-output-port compression port)))
     (dynamic-wind
       (const #f)
       (lambda ()
-        (close-port port)
         (proc compressed))
       (lambda ()
         (close-port compressed)

tests/utils.scm

@@ -164,10 +164,12 @@
 (false-if-exception (delete-file temp-file))
 (test-assert "compressed-output-port + decompressed-port"
   (let* ((file (search-path %load-path "guix/derivations.scm"))
-         (data (call-with-input-file file get-bytevector-all)))
+         (data (call-with-input-file file get-bytevector-all))
-    (call-with-compressed-output-port 'xz (open-file temp-file "w0b")
+         (port (open-file temp-file "w0b")))
+    (call-with-compressed-output-port 'xz port
       (lambda (compressed)
         (put-bytevector compressed data)))
+    (close-port port)
 
     (bytevector=? data
                   (call-with-decompressed-port 'xz (open-file temp-file "r0b")