From 6607dcb78a4c5a9ac1adb1d71c63c74b8efc120c Mon Sep 17 00:00:00 2001
From: haskal <haskal@awoo.systems>
Date: Sat, 12 Dec 2020 22:49:14 -0500
Subject: [PATCH] fix concurrency bug in comms

---
 crossfire/comms.rkt | 17 ++++++++++-------
 1 file changed, 10 insertions(+), 7 deletions(-)

diff --git a/crossfire/comms.rkt b/crossfire/comms.rkt
index 7626e6c..088e99b 100644
--- a/crossfire/comms.rkt
+++ b/crossfire/comms.rkt
@@ -107,7 +107,7 @@
 
   ;; encapsulates a message queue for one peer connection
   ;; sends and recieves encrypted data using the session key
-  (define (peer-thread peer-data session-key in out)
+  (define (peer-thread peer-data session-key in out local-msg-channel)
     (define run-queue #t)
 
     ;; handles tcp data in
@@ -121,7 +121,7 @@
                [(? msg? m)
                 (unless (= (msg-from-id m) (node-id peer-data))
                   (error "mismatched node id" (msg-from-id m) (node-id peer-data)))
-                (thread-sendrecv el-thread 'dispatch-msg (cons (node-id my-node) m))]
+                (async-channel-put local-msg-channel m)]
                [_ (error "invalid msg data from peer" (node-id peer-data))])])]
         [#f ;; likely EOF. could also be invalid fasl data that could not be deserialized
             ;; in either case, close the connection. there's not much else we can do
@@ -152,7 +152,7 @@
   ;; synchronously performs ports-proc to retrieve a new set of tcp ports, and a handshake
   ;; then starts a peer thread and a peer thread cleanup monitor
   ;; this cannot be run synchronously on the main loop because it does a thread-sendrecv
-  (define (make-peer-thread ports-proc [cust (make-custodian)])
+  (define (make-peer-thread ports-proc local-msg-channel [cust (make-custodian)])
     (define-values [new-thd peer-data session-key]
       (parameterize ([current-custodian cust])
         (define-values [in out] (ports-proc))
@@ -167,7 +167,7 @@
               (begin (engine-kill eng) (error "handshake timeout")))))
 
         (values
-          (thread (lambda () (peer-thread peer-data session-key in out)))
+          (thread (lambda () (peer-thread peer-data session-key in out local-msg-channel)))
           peer-data
           session-key)))
 
@@ -210,7 +210,7 @@
               (define (cleanup) (set! listener-thd #f))
               (with-handlers ([exn? (lambda (ex) (cleanup) (raise ex))])
                 (let loop ()
-                  (make-peer-thread (lambda () (tcp-accept listener)))
+                  (make-peer-thread (lambda () (tcp-accept listener)) local-msg-channel)
                   (loop))))))
         (thread-send from (void) #f))))
 
@@ -224,7 +224,7 @@
             [#f (thread-send from (make-error "no such node" id) #f)]
             [(node id name type pubkey seckey host port)
              (when (and host port)
-               (make-peer-thread (lambda () (tcp-connect host port))))
+               (make-peer-thread (lambda () (tcp-connect host port)) local-msg-channel))
              (thread-send from (void) #f)])))))
 
   ;; notify that startup is done
@@ -539,7 +539,10 @@
         (export #,sig-name)
         #,@(for/list ([mem (in-list members)])
             #`(define (#,mem . args)
-                (tm-transact (current-tm) (node-id (current-to-node)) (quote #,mem) args)))))
+                ;; breaks off (there's a hard timeout of 30s, so the break will be delivered
+                ;; eventually)
+                (parameterize-break #f
+                  (tm-transact (current-tm) (node-id (current-to-node)) (quote #,mem) args))))))
   unit-out)
 
 ;; this creates a wrapper unit for the given signature that delegates to the transaction manager