scheme-gnunet/tests/distributed-hash-table.scm

;; This file is part of scheme-GNUnet, a partial Scheme port of GNUnet.
;; Copyright (C) 2021, 2022 GNUnet e.V.
;;
;; scheme-GNUnet is free software: you can redistribute it and/or modify it
;; under the terms of the GNU Affero General Public License as published
;; by the Free Software Foundation, either version 3 of the License,
;; or (at your option) any later version.
;;
;; scheme-GNUnet is distributed in the hope that it will be useful, but
;; WITHOUT ANY WARRANTY; without even the implied warranty of
;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
;; Affero General Public License for more details.
;;
;; You should have received a copy of the GNU Affero General Public License
;; along with this program.  If not, see <http://www.gnu.org/licenses/>.
;;
;; SPDX-License-Identifier: AGPL-3.0-or-later
(define-module (test-distributed-hash-table))
(import (ice-9 match)
	(ice-9 binary-ports)
	(quickcheck)
	(quickcheck arbitrary)
	(quickcheck generator)
	(quickcheck property)
	(gnu gnunet dht client)
	(gnu gnunet dht network)
	(gnu gnunet dht struct)
	(gnu gnunet utils bv-slice)
	(gnu gnunet utils hat-let)
	(gnu gnunet util struct)
	(gnu gnunet netstruct syntactic)
	(only (gnu gnunet netstruct procedural) u64/big)
	(gnu gnunet hashcode struct)
	(gnu gnunet block)
	(gnu gnunet message protocols)
	(gnu gnunet mq)
	(gnu gnunet mq error-reporting)
	(gnu gnunet mq handler)
	(gnu gnunet mq-impl stream)
	(gnu extractor enum)
	(rnrs exceptions)
	(rnrs conditions)
	(rnrs base)
	(rnrs bytevectors)
	(srfi srfi-26)
	(srfi srfi-64)
	(fibers conditions)
	(fibers channels)
	(fibers operations)
	(fibers scheduler)
	(fibers timers) ; sleep
	(tests utils))

;; Use the @code{error} from Guile, not RnRS.
(define error (@ (guile) error))

;; Copied from tests/bv-slice.scm.
(define-syntax-rule (test-missing-caps test-case what permitted required code)
  (test-equal test-case
    (list what permitted required)
    (guard (c ((missing-capabilities? c)
	       (list (missing-capabilities-what c)
		     (missing-capabilities-permitted c)
		     (missing-capabilities-required c))))
      code)))

(define-syntax-rule (test-overly-large-datum test-case who canonical-type
					     length)
  (test-equal test-case
    (list who canonical-type length)
    (guard (c ((overly-large-datum? c)
	       (list (overly-large-datum-who c)
		     (missing-capabilities-permitted c)
		     (missing-capabilities-required c))))
      code)))

;; It's easy to accidentally swap the min and the max,
;; or use theoretical bounds instead of effective bounds.
(test-begin "bound-replication-level")

(define-syntax test-bound-equals
  (syntax-rules (->)
    ((_ (name argument -> expected) ...)
     (begin
       (test-equal name (list expected)
		   (call-with-values
		       (lambda ()
			 (bound-replication-level argument))
		     list))
       ...))))

(test-bound-equals
 ;; Boundaries of set of fixed points
 ("effective minimum" %effective-minimum-replication-level
  -> %effective-minimum-replication-level)
 ("effective maximum" %effective-maximum-replication-level
  -> %effective-maximum-replication-level)
 ;; off by one
 ("zero" ; remove this test if %effective-minimum-replication-level becomes zero
  (begin (assert (> %effective-minimum-replication-level %minimum-replication-level))
	 %effective-minimum-replication-level)
  -> %effective-minimum-replication-level)
 ("effective maximum + 1"
  (begin (assert (< %effective-maximum-replication-level %maximum-replication-level))
	 (+ 1 %effective-maximum-replication-level))
  -> %effective-maximum-replication-level)
 ;; Extreme values
 ("theoretical minimum" %minimum-replication-level
  -> %effective-minimum-replication-level)
 ("theoretical maximum" %maximum-replication-level
  -> %effective-maximum-replication-level))

(define between
  (map (cut + %effective-minimum-replication-level <>)
       (iota (- %effective-maximum-replication-level
		%effective-minimum-replication-level))))

;; Inner fixed points
(test-equal "between effective extrema"
  between
  (map bound-replication-level between))

(test-error "too large" (bound-replication-level (+ 1 %maximum-replication-level)))
(test-error "way too large" (bound-replication-level (* #e1e20 %maximum-replication-level)))
(test-error "too small" (bound-replication-level (- %minimum-replication-level 1)))
(test-error "way too small" (bound-replication-level (- %minimum-replication-level #e1e20)))
(test-error "non-numeric" (bound-replication-level 'what))

(define (make-slice/read-write* size)
  "Like @code{make-slice/read-write*}, but fill the slice with random data."
  (define s (make-slice/read-write size))
  (let^ ((/o/ loop (i 0))
	 (? (>= i size) s))
	(slice-u8-set! s i (random 256))
	(loop (+ i 1))))

(define* (make-a-datum #:key
		       (type 0)
		       (key (make-slice/read-write* (sizeof /hashcode:512 '())))
		       (value (make-slice/read-write 0))
		       (expiration (random (expt 2 64))))
  (make-datum type key value #:expiration expiration))
(test-assert "datum?"
  (datum? (make-a-datum)))
(test-equal "not a datum"
  '(#false #false #false)
  (map datum? (list #false 'symbol (make-slice/read-write 0))))

;; For efficiency reasons, make sure the storage is reused.
;;
;; This verifies constructing a record and extracting a field from the record
;; end ups with the value passed to the constructor, as a readable bytevector
;; slice -- the writability of the original slice, if any, is removed.
(define (slice-property-test test-case generate-slice stuff? slice->stuff stuff-slice)
  (test-assert test-case
    ;; only evaluate once, because eq? will be required
    (let* ((slice (generate-slice))
	   (stuff (slice->stuff slice))
	   (new-slice (stuff-slice stuff)))
      (and (stuff? stuff)
	   (slice-readable? new-slice)
	   (not (slice-writable? new-slice))
	   (eq? (slice-bv slice) (slice-bv new-slice))
	   (= (slice-length slice) (slice-length new-slice))))))

(define-syntax-rule (datum-key-test test-case k)
  (slice-property-test test-case (lambda () k) datum?
		       (lambda (s) (make-a-datum #:key s)) datum-key))
(define-syntax-rule (datum-value-test test-case v)
  (slice-property-test test-case (lambda () v) datum?
		       (lambda (s) (make-a-datum #:value s)) datum-value))
(define-syntax-rule (datum-type-test test-case type type/integer)
  (test-equal test-case
    type/integer
    (datum-type (make-a-datum #:type type))))

(datum-key-test "datum-key"
		(make-slice/read-write* (sizeof /hashcode:512 '())))
(datum-key-test "datum-key, read-only is sufficient"
		(slice/read-only
		 (make-slice/read-write*
		  (sizeof /hashcode:512 '()))))

(test-missing-caps
 "datum key must be readable"
 'key
 CAP_WRITE
 CAP_READ
 (make-a-datum #:key (slice/write-only (make-slice/read-write*
					(sizeof /hashcode:512 '())))))

;; AFAIK a zero length value is allowed, albeit somewhat pointless?
(datum-value-test "datum-value, length 0" (make-slice/read-write 0))
(datum-value-test "datum-value, maximal length"
		  (make-slice/read-write* %max-datum-value-length))
(datum-value-test "datum-value" (make-slice/read-write* 900))

(define (test-datum-overly-large test-case type type/integer length)
  (test-equal test-case
    (list 'make-datum type/integer length)
    (guard (c ((overly-large-datum? c)
	       (list (condition-who c)
		     (overly-large-datum-type c)
		     (overly-large-datum-length  c))))
      (make-a-datum #:type type #:value (make-slice/read-write* length)))))

(test-datum-overly-large
 "datum-value, too large (1, numeric type)" 19 19
 (* 2 %max-datum-value-length))

(test-datum-overly-large
 "datum-value, too large (2, numeric type)" 19 19
 (* 2 %max-datum-value-length))

(test-datum-overly-large
 "datum-value, too large (1, symbolic type)" (symbol-value block-type block:revocation) 12
 (* 2 %max-datum-value-length))

(datum-type-test "datum-key, symbolic type (1)"
		 (symbol-value block-type block:consensus-element) 25)
(datum-type-test "datum-key, symbolic type (2)"
		 (symbol-value block-type block:dht:hello) 7)
(datum-type-test "datum-key, known numeric type (1)" 7 7)
(datum-type-test "datum-key, known numeric type (2)" 8 8)
(datum-type-test "datum-key, unknown numeric type" 4294967295 4294967295)

(test-error "datum-type, out-of-bounds" (make-a-datum #:type 4294967296))
(test-error "datum-type, wrong enumeration"
	    (make-a-datum #:type (symbol-value message-type msg:util:dummy)))
(test-error "datum-type, wrong type (1)" (make-a-datum #:type 'foo))
;; This detected a bug!
(test-error "datum-type, wrong type (2)" (make-a-datum #:type 1.0))

(define (slice->bytevector s)
  (define b (make-bytevector (slice-length s)))
  (define s2 (bv-slice/read-write b))
  (slice-copy! s s2)
  b)

(define (query->sexp z)
  (list (query-type z) (slice->bytevector (query-key z))
	(query-desired-replication-level z)))

(define (datum->sexp z)
   (list (datum-type z)
	 (slice->bytevector (datum-key z))
	 (slice->bytevector (datum-value z))
	 (datum-expiration z)))

(define (insertion->sexp z)
  (list (datum->sexp (insertion->datum z))
	(insertion-desired-replication-level z)))

(define (search-result->sexp z)
  (list (slice->bytevector (search-result-get-path z))
	(slice->bytevector (search-result-put-path z))
	(datum->sexp (search-result->datum z))))

(define (query=? x y)
  (equal? (query->sexp x) (query->sexp y)))

(define (datum=? x y)
  (equal? (datum->sexp x) (datum->sexp y)))

(define (search-result=? x y)
  (equal? (search-result->sexp x) (search-result->sexp y)))

(define (insertion=? x y)
  (equal? (insertion->sexp x) (insertion->sexp y)))

(define (query-independent? x y)
  (slice-independent? (query-key x) (query-key y)))

(define (datum-independent? x y)
  (and (slice-independent? (datum-key x) (datum-key y))
       (slice-independent? (datum-value x) (datum-value y))))

(define (insertion-independent? x y)
  (datum-independent? (insertion->datum x) (insertion->datum y)))

(define (search-result-independent? x y)
  (and (datum-independent? (search-result->datum x) (search-result->datum y))
       (slice-independent? (search-result-get-path x)
			   (search-result-get-path y))
       (slice-independent? (search-result-put-path x)
			   (search-result-put-path y))))

(test-assert "copy-query: equal and independent"
  (let* ((old-key (make-slice/read-write* (sizeof /hashcode:512 '())))
	 (type (random 65536))
	 (desired-replication-level (+ 1 %maximum-replication-level))
	 (old (make-query type old-key))
	 (new (copy-query old)))
    (and (query=? old new)
	 (query-independent? old new))))

(test-assert "copy-datum: equal and independent"
  ;; A least in Guile 3.0.5, all bytevectors of length 0 are eq?,
  ;; so let the value be non-empty such that datum-independent?
  ;; can return #true.
  (let* ((old-key (make-slice/read-write* (sizeof /hashcode:512 '())))
	 (old-value (make-slice/read-write* 70))
	 (old (make-a-datum #:key old-key #:value old-value #:expiration 777))
	 (new (copy-datum old)))
    (and (datum=? old new)
	 (datum-independent? old new))))

(define (path-length->size l)
  (* l (sizeof /dht:path-element '())))

;; Detected a bug: the datum was not copied
(test-assert "copy-search-result: equal and independent"
  (let* ((old-key (make-slice/read-write* (sizeof /hashcode:512 '())))
	 (old-value (make-slice/read-write* 70))
	 (old-get-path (make-slice/read-write* (path-length->size 5)))
	 (old-put-path (make-slice/read-write* (path-length->size 9)))
	 (old-datum (make-a-datum #:value old-value #:expiration 555))
	 (old (datum->search-result old-datum #:get-path old-get-path
				    #:put-path old-put-path))
	 (new (copy-search-result old)))
    (and (search-result=? old new)
	 (search-result-independent? old new))))

(test-assert "copy-insertion: equal and independent"
  (let* ((old-value (make-slice/read-write* 71))
	 (old-datum (make-a-datum #:value old-value))
	 (old
	  (datum->insertion old-datum #:desired-replication-level (random 8)))
	 (new (copy-insertion old)))
    (and (insertion=? old new)
	 (insertion-independent? old new))))

(define-syntax-rule (search-result-get-path-slice-test test-case k)
  (slice-property-test test-case (lambda () k) search-result?
		       (lambda (s) (datum->search-result (make-a-datum)
							 #:get-path s))
		       search-result-get-path))
(define-syntax-rule (search-result-put-path-slice-test test-case k)
  (slice-property-test test-case (lambda () k) search-result?
		       (lambda (s) (datum->search-result (make-a-datum)
							 #:put-path s))
		       search-result-put-path))

;; These detected a bug: the capabilities were not restricted!
;; TODO: can there be a get path without a put path?
(search-result-get-path-slice-test
 "search-result-get-path, slice" (make-slice/read-write* (path-length->size 7)))
(search-result-get-path-slice-test
 "search-result-get-path, empty" (make-slice/read-write 0))
(search-result-put-path-slice-test
 "search-result-put-path, slice" (make-slice/read-write* (path-length->size 7)))
(search-result-put-path-slice-test
 "search-result-put-path, empty" (make-slice/read-write 0))
(test-equal "search-result-get-path, none"
  (list #false) ; TODO: drop 'list' when SRFI-64 bug is fixed
  (list (search-result-get-path (datum->search-result (make-a-datum)))))
(test-equal "search-result-put-path, none"
  (list #false) ; TODO: drop 'list' when SRFI-64 bug is fixed
  (list (search-result-put-path (datum->search-result (make-a-datum)))))

(test-missing-caps
 "search-result get-path must be readable"
 'get-path
 CAP_WRITE
 CAP_READ
 (datum->search-result
  (make-a-datum) #:get-path
  (slice/write-only (make-slice/read-write* (path-length->size 7)))))

(test-missing-caps
 "search-result put-path must be readable"
 'put-path
 CAP_WRITE
 CAP_READ
 (datum->search-result
  (make-a-datum) #:put-path
  (slice/write-only (make-slice/read-write* (path-length->size 7)))))

(define (test-malformed-path test-case what keyword)
  (test-assert test-case
    (quickcheck
     (property
      ((elements $natural)
       (remainder
	(arbitrary
	 (gen (choose-integer 1 (- (sizeof /dht:path-element '()) 1)))
	 (xform #false))))
      (let ((size (+ remainder (* (sizeof /dht:path-element '())))))
	(equal? (guard (c ((malformed-path? c)
			   (list (condition-who c)
				 (malformed-path-what c)
				 (malformed-path-size c))))
		  (datum->search-result
		   (make-a-datum) keyword
		   (make-slice/read-write* size)))
		(list 'datum->search-result what size)))))))

(test-malformed-path
 "get-path size must be a multiple of the size of a path element"
 'get-path #:get-path)

(test-malformed-path
 "put-path size must be a multiple of the size of a path element"
 'put-path #:put-path)


;;;
;;; Test client<->service communication.
;;;
;;; Currently, the following operations are tested:
;;;
;;;  * [x] insertion (@code{put!})
;;;  * [x] retrieval (@code{start-get!})
;;;  * [x] disconnecting
;;;  * [ ] monitoring
;;;
;;; In the following contexts:
;;;
;;;  * [x] nothing special
;;;  * [ ] after a reconnection
;;;  * [ ] requested before a reconnection, without being processed
;;;        before the reconnection.
;;;  * [ ] requested (and started) before a reconnection and continued
;;;        after the reconnection
;;;
;;; Cancelling, closing the connection, parallelism and multiple
;;; in-progress requests are currently untested (TBD and implemented!).

(define i (datum->insertion (make-a-datum) #:desired-replication-level 7))

(define (no-error-handler . e)
  (pk 'e e)
  (error "no error handler"))

;; TODO: would be nice to turn this in a real service
;; (gnu gnunet dht service).
(define* (simulate-dht-service #:optional (explode (make-condition)))
  "Simulate a DHT service, remembering all insertions and ignoring expiration
and replication.  Cancellation is ignored (TODO).  Only a single client is
supported.  When @var{explode} is signalled, the connection is closed."
  (define (slice->bv slice)
    (define bv (make-bytevector (slice-length slice)))
    (define bv/slice (bv-slice/read-write bv))
    (slice-copy! slice bv/slice)
    bv)
  (define (query->key query)
    (cons (query-type query) (slice->bv (query-key query))))
  (define (insertion->key insertion)
    (define datum (insertion->datum insertion))
    (cons (datum-type datum) (slice->bv (datum-key datum))))
  ;; Mapping from (numeric type + key bytevector)
  ;;   --> (list of value . interested mq channels)
  (define table (make-hash-table))
  (define table-channel (make-channel))
  (define mq)
  (define mq-defined (make-condition))
  (define (handle-table spawn-fiber)
    (define (put-message/async channel message)
      (assert (channel? channel))
      (spawn-fiber
       (lambda ()
	 (put-message channel message))))
    (match (perform-operation
	    (choice-operation (get-operation table-channel)
			      (wrap-operation (wait-operation explode)
					      (lambda () 'explode))))
      ('explode
       (wait mq-defined)
       (close-queue! mq))
      (('start-get! query response-channel)
       (let* ((key (query->key query))
	      (old (hash-ref table key '(() . ())))
	      (old-values (car old))
	      (channels (cdr old)))
	 ;; Send currently known values.
	 (for-each
	  (lambda (old-value)
	    (put-message/async response-channel old-value))
	  old-values)
	 ;; Send future values to the channel as well.
	 (hash-set! table key
		    `(,old-values ,response-channel ,@channels))))
      (('put! insertion)
       (let* ((key (insertion->key insertion))
	      (old (hash-ref table key '(() . ())))
	      (old-values (car old))
	      (channels (cdr old))
	      (new-values (cons insertion old-values)))
	 ;; Send the new value.
	 (for-each
	  (lambda (response-channel)
	    (put-message/async response-channel insertion))
	  channels)
	 (hash-set! table key `(,new-values . ,channels)))))
    (handle-table spawn-fiber))
  (lambda (port spawn-fiber)
    (spawn-fiber (lambda () (handle-table spawn-fiber)))
    (let^ ((! (simple-message-handler type* handle!*)
	      (message-handler
	       (type type*)
	       ((interpose foo) foo)
	       ((well-formed? s) #true)
	       ((handle! slice) (handle!* slice))))
	   (!^ (handle/put! message)
	       "Respond to a @code{/:msg:dht:client:put} message."
	       ((<-- (insertion _) (analyse-client-put message))
		(! insertion (copy-insertion insertion)))
	       (put-message table-channel `(put! ,insertion)))
	   (!^ (handle/start-get! message)
	       ""
	       ((! channel (make-channel))
		(<-- (query unique-id _) (analyse-client-get message))
		(! query (copy-query query)))
	       (put-message table-channel `(start-get! ,query ,channel))
	       (spawn-fiber
		(lambda ()
		  (let^ ((/o/ loop)
			 (! insertion (get-message channel))
			 ;; The tests don't require get-path/put-path.
			 (! search-result (datum->search-result
					   (insertion->datum insertion)))
			 (! message (construct-client-result search-result
							     unique-id)))
			(wait mq-defined)
			(send-message! mq message)
			(loop))))
	       (values))
	   (! h (message-handlers
		 (simple-message-handler
		  (symbol-value message-type msg:dht:client:put)
		  handle/put!)
		 ;; TODO: handle properly
		 (simple-message-handler
		  (symbol-value message-type msg:dht:client:get:stop)
		  (lambda (slice) (values)))
		 (simple-message-handler
		  (symbol-value message-type msg:dht:client:get)
		  handle/start-get!))))
	  (set! mq
	    (port->message-queue port h no-error-handler #:spawn spawn-fiber))
	  (signal-condition! mq-defined)
	  (values))))

(test-equal "put! sends one message to service, after connecting"
  i
  (let^ ((! connected? #false)
	 (! (connected)
	    (assert (not connected?))
	    (set! connected? #true))
	 (! message #false)
	 (! message-received (make-condition))
	 (! (handle slice)
	    (when message
	      (error "already received"))
	    (set! message slice)
	    (signal-condition! message-received))
	 (! h (message-handlers
	       (message-handler
		(type (symbol-value message-type msg:dht:client:put))
		((interpose foo) foo)
		((well-formed? s) #true)
		((handle! slice) (handle slice))))))
	(call-with-services/fibers
	 `(("dht" . ,(lambda (port spawn-fiber)
		       (define mq
			 (port->message-queue port h no-error-handler
					      #:spawn spawn-fiber))
		       (values))))
	 (lambda (config spawn-fiber)
	   (define server
	     (connect config #:connected connected #:spawn spawn-fiber))
	   (put! server i)
	   (wait message-received)
	   (pk 'server server) ; keep 'server' reachable
	   (assert connected?)
	   (assert message)
	   (let^ ((<-- (insertion _)
		       (analyse-client-put message)))
		 ;; TODO: copy to make equal? work
		 ;; (TODO: define equal? for slices)
		 (copy-insertion insertion))))))

;; Squat two message types for tests below.
(define type:ping 7)
(define type:pong 8)

(test-assert "synchronuous ping-pong with multiple balls (no interruptions, no cancellation)"
  (call-with-services/fibers
   `(("dht" . ,(simulate-dht-service)))
   (lambda (config spawn-fiber)
     (define N_ROUNDS 50)
     (define server
       (connect config #:spawn spawn-fiber))
     (define (round->key round)
       (define key (make-slice/read-write (sizeof /hashcode:512 '())))
       (slice-u64-set! key 0 round (endianness little))
       key)
     (define (make-a-insertion type round j)
       (define key (round->key round))
       (define value (make-slice/read-write 8))
       (slice-u64-set! value 0 j (endianness little))
       (datum->insertion (make-datum type key value)))
     (define (make-a-query type round)
       (define key (round->key round))
       (make-query type key))
     (define (n-responses-for-round round)
       (+ 1 (mod round 8)))
     (define (ping/pong type round)
       ;; round: number (used as key)
       ;; j: value
       ;;
       ;; Multiple values are inserted for the same key,
       ;; to test iteration.
       (let loop ((j 0))
	 (when (< j (n-responses-for-round round))
	   (put! server (make-a-insertion type round j))
	   (loop (+ 1 j)))))
     (define (search-result->j type search-result)
       (define datum (search-result->datum search-result))
       (define value (datum-value datum))
       (assert (= (slice-length value) 8)) ; u64
       (assert (= type (datum-type datum)))
       (slice-u64-ref value 0 (endianness little)))
     (define (wait-for-values type round)
       (define done (make-condition))
       (define responses '())
       (define (found search-result)
	 (set! responses
	       (cons (search-result->j type search-result) responses))
	 (define length/current (length responses))
	 (define length/expected (n-responses-for-round round))
	 (when (>= length/current length/expected)
	   ;; Duplicated responses might happen in practice, but should
	   ;; be avoided when feasible.
	   (assert (= length/current length/expected))
	   (assert (equal? (sort responses <) (iota length/expected)))
	   ;; TODO: cancel query
	   (signal-condition! done)))
       (define search (start-get! server (make-a-query type round) found
				  ;; Not testing cancellation on GC here.
				  #:linger? #true))
       (wait done))
     (define* (ping/pong* this-type other-type round)
       (when (< round N_ROUNDS)
	 (ping/pong this-type round)
	 (wait-for-values other-type round)
	 (ping/pong* this-type other-type (+ 1 round))))
     (define (spawn-ping/pong* this-type other-type)
       (define done (make-condition))
       (spawn-fiber
	(lambda ()
	  (ping/pong* this-type other-type 0)
	  (signal-condition! done)))
       done)
     (define ping (spawn-ping/pong* type:ping type:pong))
     (define pong (spawn-ping/pong* type:pong type:ping))
     (wait ping)
     (wait pong)
     #true)))

(test-assert "(DHT) close, not connected --> all fibers stop, no callbacks called"
  (close-not-connected-no-fallbacks "dht" connect disconnect!))
(test-assert "(DHT) garbage collectable"
  (garbage-collectable "dht" connect))

(define* (determine-reported-errors proc #:key (n-connections 1) (n-errors 1))
  (call-with-spawner/wait*
   (lambda (config spawn)
     (define errors '())
     (define currently-connected? #false)
     (define times-connected 0)
     (define times-errored 0)
     (define finally-disconnected-c (make-condition))
     (define all-errors-c (make-condition))
     (parameterize ((error-reporter (lambda foo
				      (assert (> times-connected 0))
				      (set! times-errored (+ 1 times-errored))
				      (set! errors (cons foo errors))
				      (when (>= times-errored n-errors)
					(signal-condition! all-errors-c)))))
       (define (connected)
	 (assert (not currently-connected?))
	 (set! currently-connected? #true)
	 (set! times-connected (+ 1 times-connected))
	 (assert (<= times-connected n-connections)))
       (define (disconnected)
	 (assert currently-connected?)
	 (set! currently-connected? #false)
	 (when (= times-connected n-connections)
	   (signal-condition! finally-disconnected-c)))
       (define server
	 (connect config #:connected connected #:disconnected disconnected
		  #:spawn spawn))
       ;; Give 'error-reporter' a chance to be called too often
       (sleep 0.001)
       ;; The error handler and 'disconnected' are called in no particular
       ;; order, so we have to wait for both.
       (wait finally-disconnected-c)
       (wait all-errors-c)
       ;; keep 'server' reachable long enough.
       (pk server)
       (and (not currently-connected?)
	    (= times-connected n-connections) errors)))
   `(("dht" . ,proc))))

(define (put-ill-formed-message port)
  (define b (make-bytevector (sizeof /:message-header '())))
  (define s (slice/write-only (bv-slice/read-write b)))
  (set%! /:message-header '(type) s
	 (value->index (symbol-value message-type msg:dht:client:result)))
  (set%! /:message-header '(size) s (slice-length s))
  (put-bytevector port b))

(test-equal "(DHT) ill-formed message from service --> all fibers stop, 'connected' and 'disconnected' called"
  `((logic:ill-formed
     ,(value->index (symbol-value message-type msg:dht:client:result))))
  (determine-reported-errors
   (lambda (port spawn-fiber)
     (put-ill-formed-message port)
     (close-port port))))

;; Allow reconnecting a few times and eventually ensure a permanent
;; disconnecting to make the test terminate.
(define n-connections 7)
(test-equal "(DHT) end-of-file --> reconnect (all fibers eventually stop)"
  `((logic:ill-formed
     ,(value->index (symbol-value message-type msg:dht:client:result))))
  (determine-reported-errors
   (let ((i 0))
     (lambda (port spawn-fiber)
       (set! i (+ i 1))
       (assert (<= i n-connections))
       (when (= i n-connections)
	 (put-ill-formed-message port))
       (close-port port)))
   #:n-connections n-connections))

;; TODO: would be nice to test that old requests are submitted again

;; The aim is to show that the search callback can start search requests
;; of its own without any problems.  While we're at it, the search results
;; are verified.
;;
;; First 'loop' searches for key 0, then for key 1 inside the search result
;; callback, etc.
(test-assert "search callback re-entrancy"
  (call-with-services/fibers
   `(("dht" . ,(simulate-dht-service)))
   (lambda (config spawn-fiber)
     (define server (connect config))
     (define ROUNDS 20)
     (define type 0) ; arbitrary
     (define (make-a-query round)
       (define key (make-slice/read-write (sizeof /hashcode:512 '())))
       (slice-u64-set! key 0 round (endianness big))
       (make-query type key))
     (define (value round)
       (expt 2 round))
     (define done (make-condition))
     (let loop ((round 0))
       (define found? #false)
       (if (< round ROUNDS)
	   (start-get! server (make-a-query round)
		       (lambda (search-result)
			 (define d (search-result->datum search-result))
			 (assert (= round
				    (slice-u64-ref (datum-key d) 0
						   (endianness big))))
			 (assert (= (value round)
				    (slice-u64-ref (datum-value d) 0
						   (endianness big))))
			 (assert (not found?))
			 (set! found? #true)
			 (loop (+ round 1)))
		       ;; Cancellation is tested elsewhere, don't automatically
		       ;; cancel.
		       #:linger? #true)
	   (signal-condition! done)))
     (let loop ((round 0))
       (define key-s (make-slice/read-write (sizeof /hashcode:512 '())))
       (define value-s (make-slice/read-write (sizeof u64/big '())))
       (slice-u64-set! key-s 0 round (endianness big))
       (slice-u64-set! value-s 0 (value round) (endianness big))
       (put! server (datum->insertion (make-datum type key-s value-s)))
       (when (< round (- ROUNDS 1))
	 (loop (+ round 1))))
     (wait done)
     #true)))

;; TODO: would be nice to verify that the necessary messages are sent to the
;; DHT service.
(test-assert "cancelling a search within a search callback does not hang"
  (call-with-services/fibers
   `(("dht" . ,(simulate-dht-service)))
   (lambda (config spawn-fiber)
     (define server (connect config))
     (define datum (make-a-datum))
     (define query (make-query (datum-type datum) (datum-key datum)))
     (define search-defined (make-condition))
     (define done (make-condition))
     (define search
       (start-get! server query (lambda (a-result)
				  (wait search-defined)
				  (stop-get! search)
				  (signal-condition! done))
		   ;; The 'found' callback is responsible for cancellation.
		   #:linger? #true))
     (signal-condition! search-defined)
     (put! server (datum->insertion datum))
     (wait done)
     #true)))

(test-assert "cancelling a search multiple times does not hang"
  (call-with-services/fibers
   `(("dht" . ,(simulate-dht-service)))
   (lambda (config spawn-fiber)
     (define server (connect config))
     (define datum (make-a-datum))
     (define query (make-query (datum-type datum) (datum-key datum)))
     (define search (start-get! server query (lambda (foo) (values))
				;; Not testing cancellation on GC here.
				#:linger? #true))
     (let loop ((n 0))
       (when (< n 40)
	 (stop-get! search)
	 (loop (+ n 1))))
     #true)))

(test-assert "searches restarted after disconnect"
  (let ((stop-first-server (make-condition))
	(first-accepted (make-condition)))
    (call-with-services/fibers
     `(("dht" . ,(lambda args
		   (if (signal-condition! first-accepted)
		       (apply (simulate-dht-service stop-first-server) args)
		       (apply (simulate-dht-service) args)))))
     (lambda (config spawn-fiber)
       (define connected/condition (make-condition))
       (define disconnected/condition (make-condition))
       (define (connected)
	 (signal-condition! connected/condition))
       (define (disconnected)
	 (signal-condition! disconnected/condition))
       (define server (connect config #:connected connected
			       #:disconnected disconnected
			       #:spawn spawn-fiber))
       ;; Start a search
       (define datum (make-a-datum))
       (define found/condition (make-condition))
       (define (found search-result)
	 (unless (datum=? datum (search-result->datum search-result))
	   (error "wrong search result"))
	 (unless (signal-condition! found/condition)
	   (error "multiple results")))
       (define query (make-query (datum-type datum) (datum-key datum)))
       (define search (start-get! server query found))
       ;; Give @var{server} a chance to actually send the request.
       ;; Removing the 'let loop' is possible, but would test some
       ;; different code paths (TODO enveloppe confirmation/cancellation).
       (wait connected/condition)
       (wait first-accepted)
       (let loop ((n 0))
	 (when (< n 100)
	   (yield-current-task)))
       ;; Break the connection, letting @var{server} reconnect.
       (signal-condition! stop-first-server)
       (wait disconnected/condition)
       ;; Insert the datum, such that @var{search} can complete (assuming
       ;; that @var{server} remembered to start the search again!).
       (put! server (datum->insertion datum))
       (wait found/condition)
       ;; Explicitely cancel 'search' such that it is not cancelled too
       ;; early due to GC.
       (stop-get! search)
       #true))))

(test-end)