6.0.2.0

* agent: collect query errors stats

* simplify

* test

* use microseconds

* parens

* revert change to track all queries, reduce threshold to 1 ms

CHANGELOG.md

@@ -1,3 +1,27 @@
+# 6.0.2
+
+SMP agent:
+- fix stuck connection commands when a server is not responding.
+- store query errors, reduce slow query threshold to 1ms.
+
+Notification server:
+- reduce PING interval to 1 minute.
+- fix subscriptions disabled on race condition (only mark subscriptions with END status when received via the active connection).
+
+# 6.0.1
+
+SMP agent:
+- support changing user of the new connection.
+- do not start delivery workers when there are no messages to deliver.
+- enable notifications for all connections.
+- combine database transactions when subscribing.
+
+SMP server:
+- safe compacting of store log.
+- fix possible race when creating client that might lead to memory leak.
+
+Dependencies: upgrade tls to 1.9
+
 # 6.0.0
 
 Version 6.0.0.8

README.md

@@ -150,7 +150,7 @@ You can install and setup servers automatically using our script:
 
 ```sh
 curl --proto '=https' --tlsv1.2 -sSf https://raw.githubusercontent.com/simplex-chat/simplexmq/stable/install.sh -o simplex-server-install.sh \
-&& if echo 'b8cf2be103f21f9461d9a500bcd3db06ab7d01d68871b07f4bd245195cbead1d simplex-server-install.sh' | sha256sum -c; then chmod +x ./simplex-server-install.sh && ./simplex-server-install.sh; rm ./simplex-server-install.sh; else echo "SHA-256 checksum is incorrect!" && rm ./simplex-server-install.sh; fi
+&& if echo 'c90886104cd640b2ed64921dba80e90691db36788e8d6dcc13d8f33f92f0ea54 simplex-server-install.sh' | sha256sum -c; then chmod +x ./simplex-server-install.sh && ./simplex-server-install.sh; rm ./simplex-server-install.sh; else echo "SHA-256 checksum is incorrect!" && rm ./simplex-server-install.sh; fi
 ```
 
 ### Build from source

install.sh

@@ -3,8 +3,7 @@ set -eu
 
 # Links to scripts/configs
 bin="https://github.com/simplex-chat/simplexmq/releases/latest/download"
-bin_smp="$bin/smp-server-ubuntu-20_04-x86-64"
-bin_xftp="$bin/xftp-server-ubuntu-20_04-x86-64"
+remote_version="$(curl --proto '=https' --tlsv1.2 -sSf -L https://api.github.com/repos/simplex-chat/simplexmq/releases/latest | grep -i "tag_name" | awk -F \" '{print $4}')"
 
 scripts="https://raw.githubusercontent.com/simplex-chat/simplexmq/stable/scripts/main"
 scripts_systemd_smp="$scripts/smp-server.service"
@@ -26,6 +25,8 @@ path_conf_var="/var/opt"
 path_conf_smp="$path_conf_etc/simplex $path_conf_var/simplex"
 path_conf_xftp="$path_conf_etc/simplex-xftp $path_conf_var/simplex-xftp /srv/xftp"
 
+path_conf_info="$path_conf_etc/simplex-info"
+
 path_systemd="/etc/systemd/system"
 path_systemd_smp="$path_systemd/smp-server.service"
 path_systemd_xftp="$path_systemd/xftp-server.service"
@@ -65,7 +66,13 @@ ${GRN}4.${NC} Create systemd services:
 ${GRN}5.${NC} Install stopscript (systemd), update and uninstallation script:
     - all: ${YLW}${path_bin_update}${NC}, ${YLW}${path_bin_uninstall}${NC}, ${YLW}${path_bin_stopscript}${NC}
 
-Press ${GRN}ENTER${NC} to continue or ${RED}Ctrl+C${NC} to cancel installation"
+Press:
+  - ${GRN}ENTER${NC} to continue installing both xftp and smp servers
+  - ${GRN}1${NC} to install only smp server
+  - ${GRN}2${NC} to install only xftp server
+  - ${RED}Ctrl+C${NC} to cancel installation
+  
+Selection: "
 
 end="Installtion is complete!
 
@@ -76,27 +83,64 @@ Please checkout our server guides:
 To uninstall with full clean-up, simply run: ${YLW}sudo /usr/local/bin/simplex-servers-uninstall${NC}
 "
 
+os_test() {
+ . /etc/os-release
+
+ case "$VERSION_ID" in
+  20.04|22.04) : ;;
+  24.04) VERSION_ID='22.04' ;;
+  *) printf "${RED}Unsupported Ubuntu version!${NC}\nPlease file Github issue with request to support Ubuntu %s: https://github.com/simplex-chat/simplexmq/issues/new\n" "$VERSION_ID" && exit 1 ;;
+ esac
+
+ version="$(printf '%s' "$VERSION_ID" | tr '.' '_')"
+ arch="$(uname -p)"
+
+ case "$arch" in
+  x86_64) arch="$(printf '%s' "$arch" | tr '_' '-')" ;;
+  *)  printf "${RED}Unsupported architecture!${NC}\nPlease file Github issue with request to support %s architecture: https://github.com/simplex-chat/simplexmq/issues/new" "$arch" && exit 1 ;;
+ esac
+
+ bin_smp="$bin/smp-server-ubuntu-${version}-${arch}"
+ bin_xftp="$bin/xftp-server-ubuntu-${version}-${arch}"
+}
+
 setup_bins() {
- curl --proto '=https' --tlsv1.2 -sSf -L "$bin_smp" -o "$path_bin_smp" && chmod +x "$path_bin_smp"
- curl --proto '=https' --tlsv1.2 -sSf -L "$bin_xftp" -o "$path_bin_xftp" && chmod +x "$path_bin_xftp"
+ eval "bin=\$bin_${1}"
+ eval "path=\$path_bin_${1}"
+
+ curl --proto '=https' --tlsv1.2 -sSf -L "$bin" -o "$path" && chmod +x "$path"
+
+ unset bin path
 }
 
 setup_users() {
- useradd -M "$user_smp" 2> /dev/null || true
- useradd -M "$user_xftp" 2> /dev/null || true
+ eval "user=\$user_${1}"
+
+ useradd -M "$user" 2> /dev/null || true
+
+ unset user
 }
 
 setup_dirs() {
  # Unquoted varibles, so field splitting can occur
- mkdir -p $path_conf_smp
- chown "$user_smp":"$user_smp" $path_conf_smp
- mkdir -p $path_conf_xftp
- chown "$user_xftp":"$user_xftp" $path_conf_xftp
+ eval "path_conf=\$path_conf_${1}"
+ eval "user=\$user_${1}"
+
+ mkdir -p $path_conf
+ mkdir -p $path_conf_info
+ printf "local_version_%s='%s'\n" "$1" "$remote_version" >> "$path_conf_info/release"
+ chown -R "$user":"$user" $path_conf
+
+ unset path_conf user
 }
 
 setup_systemd() {
- curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_systemd_smp" -o "$path_systemd_smp"
- curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_systemd_xftp" -o "$path_systemd_xftp"
+ eval "scripts_systemd=\$scripts_systemd_${1}"
+ eval "path_systemd=\$path_systemd_${1}"
+
+ curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_systemd" -o "$path_systemd"
+
+ unset scripts_systemd path_systemd
 }
 
 setup_scripts() {
@@ -110,32 +154,62 @@ checks() {
   printf "This script is intended to be run with root privileges. Please re-run script using sudo."
   exit 1
  fi
+ 
+ os_test
+
+ mkdir -p $path_conf_info  
 }
 
 main() {
  checks
  
- printf "%b\n%b\n" "${BLU}$logo${NC}" "$welcome"
+ printf "%b\n%b" "${BLU}$logo${NC}" "$welcome"
  read ans
 
+ if [ "$ans" = '1' ]; then
+  setup='smp'
+ elif [ "$ans" = '2' ]; then
+  setup='xftp'
+ else
+  setup='smp xftp'
+ fi
+
  printf "Installing binaries..."
- setup_bins
+ 
+ for i in $setup; do
+  setup_bins "$i"
+ done
+ 
  printf "${GRN} Done!${NC}\n"
 
  printf "Creating users..."
- setup_users
+
+ for i in $setup; do
+  setup_users "$i"
+ done
+
  printf "${GRN} Done!${NC}\n"
  
  printf "Creating directories..."
- setup_dirs
+
+ for i in $setup; do
+  setup_dirs "$i"
+ done
+
  printf "${GRN} Done!${NC}\n"
 
  printf "Creating systemd services..."
- setup_systemd
+
+ for i in $setup; do
+  setup_systemd "$i"
+ done
+
  printf "${GRN} Done!${NC}\n"
 
  printf "Installing stopscript, update and uninstallation script..."
+
  setup_scripts
+
  printf "${GRN} Done!${NC}\n"
 
  printf "%b" "$end"

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 6.0.0.8
+version: 6.0.2.0
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -69,7 +69,7 @@ dependencies:
   - temporary == 1.3.*
   - time == 1.12.*
   - time-manager == 0.0.*
-  - tls >= 1.7.0 && < 1.8
+  - tls >= 1.9.0 && < 1.10
   - transformers == 0.6.*
   - unliftio == 0.2.*
   - unliftio-core == 0.2.*

scripts/main/simplex-servers-uninstall

@@ -13,11 +13,15 @@ fi
 printf "${RED}This action will permanently remove all configs, directories, binaries from Installation Script. Please backup any relevant configs if they are needed.${NC}\n\nPress ${GRN}ENTER${NC} to continue or ${RED}Ctrl+C${NC} to cancel installation"
 read ans
 
-systemctl stop smp-server
-systemctl stop xftp-server
+systemctl disable --now smp-server 2>/dev/null || true
+systemctl revert smp-server 2>/dev/null || true
+systemctl disable --now xftp-server 2>/dev/null || true
+systemctl revert xftp-server 2>/dev/null || true
+systemctl daemon-reload 2>/dev/null || true
 
-rm -rf /var/opt/simplex /etc/opt/simplex /var/opt/simplex-xftp /etc/opt/simplex-xftp /srv/xftp /etc/systemd/system/smp-server.service /etc/systemd/system/xftp-server.service /usr/local/bin/smp-server /usr/local/bin/xftp-server /usr/local/bin/simplex-servers-update /usr/local/bin/simplex-servers-uninstall
+rm -rf /var/opt/simplex /etc/opt/simplex /etc/opt/simplex-info /var/opt/simplex-xftp /etc/opt/simplex-xftp /srv/xftp /etc/systemd/system/smp-server.service /etc/systemd/system/xftp-server.service /usr/local/bin/smp-server /usr/local/bin/xftp-server /usr/local/bin/simplex-servers-update /usr/local/bin/simplex-servers-uninstall /usr/local/bin/simplex-servers-stopscript
 
-userdel smp && userdel xftp
+userdel smp 2>/dev/null || true
+userdel xftp 2>/dev/null || true
 
 printf "Uninstallation is complete! Thanks for trying out SimpleX!\n"

scripts/main/simplex-servers-update

@@ -3,8 +3,6 @@ set -eu
 
 # Links to scripts/configs
 bin="https://github.com/simplex-chat/simplexmq/releases/latest/download"
-bin_smp="$bin/smp-server-ubuntu-20_04-x86-64"
-bin_xftp="$bin/xftp-server-ubuntu-20_04-x86-64"
 
 scripts="https://raw.githubusercontent.com/simplex-chat/simplexmq/stable/scripts/main"
 scripts_systemd_smp="$scripts/smp-server.service"
@@ -33,6 +31,9 @@ path_tmp_bin_stopscript="$path_tmp_bin/simplex-servers-stopscript"
 path_tmp_systemd_smp="$path_tmp_bin/smp-server.service"
 path_tmp_systemd_xftp="$path_tmp_bin/xftp-server.service"
 
+path_conf_etc='/etc/opt'
+path_conf_info='/etc/opt/simplex-info'
+
 GRN='\033[0;32m'
 BLU='\033[1;36m'
 YLW='\033[1;33m'
@@ -40,8 +41,40 @@ RED='\033[0;31m'
 NC='\033[0m'
 
 # Currently, XFTP default to v0.1.0, so it doesn't make sense to check its version
-local_version="$($path_bin_smp -v | awk '{print $3}')"
-remote_version="$(curl --proto '=https' --tlsv1.2 -sSf -L https://api.github.com/repos/simplex-chat/simplexmq/releases/latest | grep -i "tag_name" | awk -F \" '{print $4}')"
+
+os_test() {
+ . /etc/os-release
+
+ case "$VERSION_ID" in
+  20.04|22.04) : ;;
+  24.04) VERSION_ID='22.04' ;;
+  *) printf "${RED}Unsupported Ubuntu version!${NC}\nPlease file Github issue with request to support Ubuntu %s: https://github.com/simplex-chat/simplexmq/issues/new\n" "$VERSION_ID" && exit 1 ;;
+ esac
+
+ version="$(printf '%s' "$VERSION_ID" | tr '.' '_')"
+ arch="$(uname -p)"
+
+ case "$arch" in
+  x86_64) arch="$(printf '%s' "$arch" | tr '_' '-')" ;;
+  *)  printf "${RED}Unsupported architecture!${NC}\nPlease file Github issue with request to support %s architecture: https://github.com/simplex-chat/simplexmq/issues/new" "$arch" && exit 1 ;;
+ esac
+
+ bin_smp="$bin/smp-server-ubuntu-${version}-${arch}"
+ bin_xftp="$bin/xftp-server-ubuntu-${version}-${arch}"
+}
+
+installed_test() {
+  set +u
+  for i in $path_conf_etc/*; do
+    if [ -d "$i" ]; then
+      case "$i" in
+        *simplex) apps="smp $apps" ;;
+        *simplex-xftp) apps="xftp $apps" ;;
+      esac
+    fi
+  done
+  set -u
+}
 
 update_scripts() {
   curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_update" -o "$path_tmp_bin_update" && chmod +x "$path_tmp_bin_update"
@@ -56,6 +89,7 @@ update_scripts() {
     mv "$path_tmp_bin_uninstall" "$path_bin_uninstall"
     printf "${GRN}Done!${NC}\n"
   fi
+
   if diff -q "$path_bin_stopscript" "$path_tmp_bin_stopscript" > /dev/null 2>&1; then
     printf -- "- ${YLW}Stopscript script is up-to-date${NC}.\n"
     rm "$path_tmp_bin_stopscript"
@@ -64,6 +98,7 @@ update_scripts() {
     mv "$path_tmp_bin_stopscript" "$path_bin_stopscript"
     printf "${GRN}Done!${NC}\n"
   fi
+
   if diff -q "$path_bin_update" "$path_tmp_bin_update" > /dev/null 2>&1; then
     printf -- "- ${YLW}Update script is up-to-date${NC}.\n"
     rm "$path_tmp_bin_update"
@@ -71,75 +106,84 @@ update_scripts() {
     printf -- "- Updating update script..."
     mv "$path_tmp_bin_update" "$path_bin_update"
     printf "${GRN}Done!${NC}\n"
-    printf "Re-executing Update script with latest updates..."
+    printf -- "- Re-executing Update script with latest updates..."
     exec sh "$path_bin_update" "continue"
   fi
 }
 
 update_systemd() {
-  curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_systemd_smp" -o "$path_tmp_systemd_smp"
-  curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_systemd_xftp" -o "$path_tmp_systemd_xftp"
+  service="${1}-server"
+  eval "scripts_systemd=\$scripts_systemd_${1}"
+  eval "path_systemd=\$path_systemd_${1}"
+  eval "path_tmp_systemd=\$path_tmp_systemd_${1}"
+   
+  curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_systemd" -o "$path_tmp_systemd"
 
-  if diff -q "$path_systemd_smp" "$path_tmp_systemd_smp" > /dev/null 2>&1; then
-    printf -- "- ${YLW}smp-server service is up-to-date${NC}.\n"
-    rm "$path_tmp_systemd_smp"
+  if diff -q "$path_systemd" "$path_tmp_systemd" > /dev/null 2>&1; then
+    printf -- "- ${YLW}%s service is up-to-date${NC}.\n" "$service"
+    rm "$path_tmp_systemd"
   else
-    printf -- "- Updating smp-server service..."
-    mv "$path_tmp_systemd_smp" "$path_systemd_smp"
-    systemctl daemon-reload
-    printf "${GRN}Done!${NC}\n"
-  fi
-  if diff -q "$path_systemd_xftp" "$path_tmp_systemd_xftp" > /dev/null 2>&1; then
-    printf -- "- ${YLW}xftp-server service is up-to-date${NC}.\n"
-    rm "$path_tmp_systemd_xftp"
-  else
-    printf -- "- Updating xftp-server service..."
-    mv "$path_tmp_systemd_xftp" "$path_systemd_xftp"
+    printf -- "- Updating %s service..." "$service"
+    mv "$path_tmp_systemd" "$path_systemd"
     systemctl daemon-reload
     printf "${GRN}Done!${NC}\n"
   fi
+
+  unset service scripts_systemd path_systemd path_tmp_systemd
 }
 
 update_bins() {
-  if [ "$local_version" != "$remote_version" ]; then
-    if systemctl is-active --quiet smp-server; then
-      printf -- "- Stopping smp-server service..."
-      systemctl stop smp-server
-      printf "${GRN}Done!${NC}\n"
+  service="${1}-server"
+  eval "bin=\$bin_${1}"
+  eval "path_bin=\$path_bin_${1}"
 
-      printf -- "- Updating smp-server bin to %s..." "$remote_version"
-      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_smp" -o "$path_bin_smp" && chmod +x "$path_bin_smp"
-      printf "${GRN}Done!${NC}\n"
-      
-      printf -- "- Starting smp-server service..."
-      systemctl start smp-server
-      printf "${GRN}Done!${NC}\n"
-    else
-      printf -- "- Updating smp-server bin..."
-      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_smp" -o "$path_bin_smp" && chmod +x "$path_bin_smp"
-      printf "${GRN}Done!${NC}\n"
+  remote_version="$(curl --proto '=https' --tlsv1.2 -sSf -L https://api.github.com/repos/simplex-chat/simplexmq/releases/latest | grep -i "tag_name" | awk -F \" '{print $4}')"
+
+  set_ver() {
+    local_version='unset'
+    sed -i -- "s/local_version_${1}=.*/local_version_${1}='${remote_version}'/" "$path_conf_info/release"
+  }
+
+  if [ -f "$path_conf_info/release" ]; then
+    . "$path_conf_info/release" 2>/dev/null
+
+    set +u
+    eval "local_version=\$local_version_${1}"
+    set -u
+
+    if [ -z "${local_version}" ]; then
+            set_ver "$1"
     fi
+  else
+    printf 'local_version_xftp=\nlocal_version_smp=\n' > "$path_conf_info/release"
+    set_ver "$1"
+  fi
 
-    if systemctl is-active --quiet xftp-server; then
-      printf -- "- Stopping xftp-server service..."
-      systemctl stop xftp-server
+  if [ "$local_version" != "$remote_version" ]; then
+    if systemctl is-active --quiet "$service"; then
+      printf -- "- Stopping %s service..." "$service"
+      systemctl stop "$service"
       printf "${GRN}Done!${NC}\n"
 
-      printf -- "- Updating xftp-server bin to %s..." "$remote_version"
-      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_xftp" -o "$path_bin_xftp" && chmod +x "$path_bin_xftp"
+      printf -- "- Updating %s to %s..." "$service" "$remote_version"
+      curl --proto '=https' --tlsv1.2 -sSf -L "$bin" -o "$path_bin" && chmod +x "$path_bin"
       printf "${GRN}Done!${NC}\n"
       
-      printf -- "- Starting xftp-server service..."
-      systemctl start xftp-server
+      printf -- "- Starting %s service..." "$service"
+      systemctl start "$service"
       printf "${GRN}Done!${NC}\n"
     else
-      printf -- "- Updating xftp-server bin..."
-      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_smp" -o "$path_bin_xftp" && chmod +x "$path_bin_xftp"
+      printf -- "- Updating %s to %s..." "$service" "$remote_version"
+      curl --proto '=https' --tlsv1.2 -sSf -L "$bin" -o "$path_bin" && chmod +x "$path_bin"
       printf "${GRN}Done!${NC}\n"
     fi
   else
-    printf -- "- ${YLW}smp-server and xftp-server binaries is up-to-date${NC}.\n"
+    printf -- "- ${YLW}%s is up-to-date${NC}.\n" "$service"
   fi
+
+  set_ver "$1"
+
+  unset service bin path_bin local_version
 }
 
 checks() {
@@ -147,6 +191,11 @@ checks() {
     printf "This script is intended to be run with root privileges. Please re-run script using sudo.\n"
     exit 1
   fi
+
+  os_test
+  installed_test
+
+  mkdir -p $path_conf_info
 }
 
 main() {
@@ -163,10 +212,14 @@ main() {
   fi
 
   printf "Updating systemd services...\n"
-  update_systemd
+  for i in $apps; do
+    update_systemd "$i"
+  done
   
-  printf "Updating simplex server binaries...\n"
-  update_bins
+  printf "Updating simplex servers...\n"
+  for i in $apps; do
+    update_bins "$i"
+  done
 
   rm -rf "$path_tmp_bin"
 }

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        6.0.0.8
+version:        6.0.2.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -258,7 +258,7 @@ library
     , temporary ==1.3.*
     , time ==1.12.*
     , time-manager ==0.0.*
-    , tls >=1.7.0 && <1.8
+    , tls >=1.9.0 && <1.10
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
@@ -333,7 +333,7 @@ executable ntf-server
     , temporary ==1.3.*
     , time ==1.12.*
     , time-manager ==0.0.*
-    , tls >=1.7.0 && <1.8
+    , tls >=1.9.0 && <1.10
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
@@ -412,7 +412,7 @@ executable smp-server
     , temporary ==1.3.*
     , time ==1.12.*
     , time-manager ==0.0.*
-    , tls >=1.7.0 && <1.8
+    , tls >=1.9.0 && <1.10
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
@@ -490,7 +490,7 @@ executable xftp
     , temporary ==1.3.*
     , time ==1.12.*
     , time-manager ==0.0.*
-    , tls >=1.7.0 && <1.8
+    , tls >=1.9.0 && <1.10
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
@@ -565,7 +565,7 @@ executable xftp-server
     , temporary ==1.3.*
     , time ==1.12.*
     , time-manager ==0.0.*
-    , tls >=1.7.0 && <1.8
+    , tls >=1.9.0 && <1.10
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
@@ -681,7 +681,7 @@ test-suite simplexmq-test
     , time ==1.12.*
     , time-manager ==0.0.*
     , timeit ==2.0.*
-    , tls >=1.7.0 && <1.8
+    , tls >=1.9.0 && <1.10
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*

src/Simplex/Messaging/Agent.hs

@@ -53,6 +53,7 @@ module Simplex.Messaging.Agent
     deleteConnectionAsync,
     deleteConnectionsAsync,
     createConnection,
+    changeConnectionUser,
     prepareConnectionToJoin,
     joinConnection,
     allowConnection,
@@ -127,7 +128,7 @@ import Control.Monad.Reader
 import Control.Monad.Trans.Except
 import Crypto.Random (ChaChaDRG)
 import qualified Data.Aeson as J
-import Data.Bifunctor (bimap, first)
+import Data.Bifunctor (bimap, first, second)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Composition ((.:), (.:.), (.::), (.::.))
@@ -333,6 +334,11 @@ createConnection :: AgentClient -> UserId -> Bool -> SConnectionMode c -> Maybe
 createConnection c userId enableNtfs = withAgentEnv c .:: newConn c userId "" enableNtfs
 {-# INLINE createConnection #-}
 
+-- | Changes the user id associated with a connection 
+changeConnectionUser :: AgentClient -> UserId -> ConnId -> UserId -> AE ()
+changeConnectionUser c oldUserId connId newUserId = withAgentEnv c $ changeConnectionUser' c oldUserId connId newUserId
+{-# INLINE changeConnectionUser #-}
+
 -- | Create SMP agent connection without queue (to be joined with joinConnection passing connection ID).
 -- This method is required to prevent race condition when confirmation from peer is received before
 -- the caller of joinConnection saves connection ID to the database.
@@ -741,6 +747,16 @@ newConn :: AgentClient -> UserId -> ConnId -> Bool -> SConnectionMode c -> Maybe
 newConn c userId connId enableNtfs cMode clientData pqInitKeys subMode =
   getSMPServer c userId >>= newConnSrv c userId connId False enableNtfs cMode clientData pqInitKeys subMode
 
+changeConnectionUser' :: AgentClient -> UserId -> ConnId -> UserId -> AM ()
+changeConnectionUser' c oldUserId connId newUserId = do
+  SomeConn _ conn <- withStore c (`getConn` connId)
+  case conn of
+    NewConnection {} -> updateConn
+    RcvConnection {} -> updateConn
+    _ -> throwE $ CMD PROHIBITED "changeConnectionUser: established connection"
+  where
+    updateConn = withStore' c $ \db -> setConnUserId db oldUserId connId newUserId
+
 newConnSrv :: AgentClient -> UserId -> ConnId -> Bool -> Bool -> SConnectionMode c -> Maybe CRClientData -> CR.InitialKeys -> SubscriptionMode -> SMPServerWithAuth -> AM (ConnId, ConnectionRequestUri c)
 newConnSrv c userId connId hasNewConn enableNtfs cMode clientData pqInitKeys subMode srv = do
   connId' <-
@@ -958,12 +974,12 @@ subscribeConnections' c connIds = do
   let (errs, cs) = M.mapEither id conns
       errs' = M.map (Left . storeError) errs
       (subRs, rcvQs) = M.mapEither rcvQueueOrResult cs
-  mapM_ (mapM_ (\(cData, sqs) -> mapM_ (lift . resumeMsgDelivery c cData) sqs) . sndQueue) cs
-  mapM_ (resumeConnCmds c) $ M.keys cs
+  resumeDelivery cs
+  lift $ resumeConnCmds c $ M.keys cs
   rcvRs <- lift $ connResults . fst <$> subscribeQueues c (concat $ M.elems rcvQs)
   ns <- asks ntfSupervisor
   tkn <- readTVarIO (ntfTkn ns)
-  when (instantNotifications tkn) . void . lift . forkIO . void . runExceptT $ sendNtfCreate ns rcvRs conns
+  lift $ when (instantNotifications tkn) . void . forkIO . void $ sendNtfCreate ns rcvRs cs
   let rs = M.unions ([errs', subRs, rcvRs] :: [Map ConnId (Either AgentErrorType ())])
   notifyResultError rs
   pure rs
@@ -995,15 +1011,20 @@ subscribeConnections' c connIds = do
         order (Active, _) = 2
         order (_, Right _) = 3
         order _ = 4
-    sendNtfCreate :: NtfSupervisor -> Map ConnId (Either AgentErrorType ()) -> Map ConnId (Either StoreError SomeConn) -> AM ()
-    sendNtfCreate ns rcvRs conns =
-      forM_ (M.assocs rcvRs) $ \case
-        (connId, Right _) -> forM_ (M.lookup connId conns) $ \case
-          Right (SomeConn _ conn) -> do
-            let cmd = if enableNtfs $ toConnData conn then NSCCreate else NSCDelete
-            atomically $ writeTBQueue (ntfSubQ ns) (connId, cmd)
-          _ -> pure ()
-        _ -> pure ()
+    sendNtfCreate :: NtfSupervisor -> Map ConnId (Either AgentErrorType ()) -> Map ConnId SomeConn -> AM' ()
+    sendNtfCreate ns rcvRs cs = do
+      -- TODO this needs to be batched end to end.
+      -- Currently, the only change is to ignore failed subscriptions.
+      let oks = M.keysSet $ M.filter (either temporaryAgentError $ const True) rcvRs
+      forM_ (M.restrictKeys cs oks) $ \case
+        SomeConn _ conn -> do
+          let cmd = if enableNtfs $ toConnData conn then NSCCreate else NSCDelete
+              ConnData {connId} = toConnData conn
+          atomically $ writeTBQueue (ntfSubQ ns) (connId, cmd)            
+    resumeDelivery :: Map ConnId SomeConn -> AM ()
+    resumeDelivery conns = do
+      conns' <- M.restrictKeys conns . S.fromList <$> withStore' c getConnectionsForDelivery
+      lift $ mapM_ (mapM_ (\(cData, sqs) -> mapM_ (resumeMsgDelivery c cData) sqs) . sndQueue) conns'
     sndQueue :: SomeConn -> Maybe (ConnData, NonEmpty SndQueue)
     sndQueue (SomeConn _ conn) = case conn of
       DuplexConnection cData _ sqs -> Just (cData, sqs)
@@ -1112,40 +1133,39 @@ sendMessagesB_ c reqs connIds = withConnLocks c connIds "sendMessages" $ do
 enqueueCommand :: AgentClient -> ACorrId -> ConnId -> Maybe SMPServer -> AgentCommand -> AM ()
 enqueueCommand c corrId connId server aCommand = do
   withStore c $ \db -> createCommand db corrId connId server aCommand
-  lift . void $ getAsyncCmdWorker True c server
+  lift . void $ getAsyncCmdWorker True c connId server
 
-resumeSrvCmds :: AgentClient -> Maybe SMPServer -> AM' ()
-resumeSrvCmds = void .: getAsyncCmdWorker False
+resumeSrvCmds :: AgentClient -> ConnId -> Maybe SMPServer -> AM' ()
+resumeSrvCmds = void .:. getAsyncCmdWorker False
 {-# INLINE resumeSrvCmds #-}
 
-resumeConnCmds :: AgentClient -> ConnId -> AM ()
-resumeConnCmds c connId =
-  unlessM connQueued $
-    withStore' c (`getPendingCommandServers` connId)
-      >>= mapM_ (lift . resumeSrvCmds c)
-  where
-    connQueued = atomically $ isJust <$> TM.lookupInsert connId True (connCmdsQueued c)
+resumeConnCmds :: AgentClient -> [ConnId] -> AM' ()
+resumeConnCmds c connIds = do
+  connSrvs <- rights . zipWith (second . (,)) connIds <$> withStoreBatch' c (\db -> fmap (getPendingCommandServers db) connIds)
+  mapM_ (\(connId, srvs) -> mapM_ (resumeSrvCmds c connId) srvs) connSrvs
 
-getAsyncCmdWorker :: Bool -> AgentClient -> Maybe SMPServer -> AM' Worker
-getAsyncCmdWorker hasWork c server =
-  getAgentWorker "async_cmd" hasWork c server (asyncCmdWorkers c) (runCommandProcessing c server)
+getAsyncCmdWorker :: Bool -> AgentClient -> ConnId -> Maybe SMPServer -> AM' Worker
+getAsyncCmdWorker hasWork c connId server =
+  getAgentWorker "async_cmd" hasWork c (connId, server) (asyncCmdWorkers c) (runCommandProcessing c connId server)
 
-runCommandProcessing :: AgentClient -> Maybe SMPServer -> Worker -> AM ()
-runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do
+data CommandCompletion = CCMoved | CCCompleted
+
+runCommandProcessing :: AgentClient -> ConnId -> Maybe SMPServer -> Worker -> AM ()
+runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
   ri <- asks $ messageRetryInterval . config -- different retry interval?
   forever $ do
     atomically $ endAgentOperation c AOSndNetwork
     lift $ waitForWork doWork
     liftIO $ throwWhenInactive c
     atomically $ beginAgentOperation c AOSndNetwork
-    withWork c doWork (`getPendingServerCommand` server_) $ runProcessCmd (riFast ri)
+    withWork c doWork (\db -> getPendingServerCommand db connId server_) $ runProcessCmd (riFast ri)
   where
     runProcessCmd ri cmd = do
       pending <- newTVarIO []
       processCmd ri cmd pending
       mapM_ (atomically . writeTBQueue subQ) . reverse =<< readTVarIO pending
     processCmd :: RetryInterval -> PendingCommand -> TVar [ATransmission] -> AM ()
-    processCmd ri PendingCommand {cmdId, corrId, userId, connId, command} pendingCmds = case command of
+    processCmd ri PendingCommand {cmdId, corrId, userId, command} pendingCmds = case command of
       AClientCommand cmd -> case cmd of
         NEW enableNtfs (ACM cMode) pqEnc subMode -> noServer $ do
           usedSrvs <- newTVarIO ([] :: [SMPServer])
@@ -1171,16 +1191,27 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do
       AInternalCommand cmd -> case cmd of
         ICAckDel rId srvMsgId msgId -> withServer $ \srv -> tryWithLock "ICAckDel" $ ack srv rId srvMsgId >> withStore' c (\db -> deleteMsg db connId msgId)
         ICAck rId srvMsgId -> withServer $ \srv -> tryWithLock "ICAck" $ ack srv rId srvMsgId
-        ICAllowSecure _rId senderKey -> withServer' . tryWithLock "ICAllowSecure" $ do
+        ICAllowSecure _rId senderKey -> withServer' . tryMoveableWithLock "ICAllowSecure" $ do
           (SomeConn _ conn, AcceptedConfirmation {senderConf, ownConnInfo}) <-
             withStore c $ \db -> runExceptT $ (,) <$> ExceptT (getConn db connId) <*> ExceptT (getAcceptedConfirmation db connId)
           case conn of
             RcvConnection cData rq -> do
               mapM_ (secure rq) senderKey
               mapM_ (connectReplyQueues c cData ownConnInfo Nothing) (L.nonEmpty $ smpReplyQueues senderConf)
+              pure CCCompleted
             -- duplex connection is matched to handle SKEY retries
-            DuplexConnection cData _ (sq :| _) ->
-              mapM_ (connectReplyQueues c cData ownConnInfo (Just sq)) (L.nonEmpty $ smpReplyQueues senderConf)
+            DuplexConnection cData _ (sq :| _) -> do
+              tryAgentError (mapM_ (connectReplyQueues c cData ownConnInfo (Just sq)) (L.nonEmpty $ smpReplyQueues senderConf)) >>= \case
+                Right () -> pure CCCompleted
+                Left e
+                  | temporaryOrHostError e && Just server /= server_ -> do
+                      -- In case the server is different we update server to remove command from this (connId, srv) queue
+                      withStore c $ \db -> updateCommandServer db cmdId server
+                      lift . void $ getAsyncCmdWorker True c connId (Just server)
+                      pure CCMoved
+                  | otherwise -> throwE e
+              where
+                server = qServer sq
             _ -> throwE $ INTERNAL $ "incorrect connection type " <> show (internalCmdTag cmd)
         ICDuplexSecure _rId senderKey -> withServer' . tryWithLock "ICDuplexSecure" . withDuplexConn $ \(DuplexConnection cData (rq :| _) (sq :| _)) -> do
           secure rq senderKey
@@ -1253,15 +1284,18 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do
           withStore c (`getConn` connId) >>= \case
             SomeConn _ conn@DuplexConnection {} -> a conn
             _ -> internalErr "command requires duplex connection"
-        tryCommand action = withRetryInterval ri $ \_ loop -> do
+        tryCommand action = tryMoveableCommand (action $> CCCompleted)
+        tryMoveableCommand action = withRetryInterval ri $ \_ loop -> do
           liftIO $ waitWhileSuspended c
           liftIO $ waitForUserNetwork c
-          tryError action >>= \case
+          tryAgentError action >>= \case
             Left e
               | temporaryOrHostError e -> retrySndOp c loop
               | otherwise -> cmdError e
-            Right () -> withStore' c (`deleteCommand` cmdId)
+            Right CCCompleted -> withStore' c (`deleteCommand` cmdId)
+            Right CCMoved -> pure () -- command processing moved to another command queue
         tryWithLock name = tryCommand . withConnLock c connId name
+        tryMoveableWithLock name = tryMoveableCommand . withConnLock c connId name
         internalErr s = cmdError $ INTERNAL $ s <> ": " <> show (agentCommandTag command)
         cmdError e = notify (ERR e) >> withStore' c (`deleteCommand` cmdId)
         notify :: forall e. AEntityI e => AEvent e -> AM ()

src/Simplex/Messaging/Agent/Client.hs

@@ -163,7 +163,7 @@ module Simplex.Messaging.Agent.Client
 where
 
 import Control.Applicative ((<|>))
-import Control.Concurrent (ThreadId, forkIO)
+import Control.Concurrent (ThreadId, killThread)
 import Control.Concurrent.Async (Async, uninterruptibleCancel)
 import Control.Concurrent.STM (retry)
 import Control.Exception (AsyncException (..), BlockedIndefinitelyOnSTM (..))
@@ -266,10 +266,11 @@ import Simplex.Messaging.Transport (SMPVersion, SessionId, THandleParams (sessio
 import Simplex.Messaging.Transport.Client (TransportHost (..))
 import Simplex.Messaging.Util
 import Simplex.Messaging.Version
-import System.Mem.Weak (Weak)
+import System.Mem.Weak (Weak, deRefWeak)
 import System.Random (randomR)
 import UnliftIO (mapConcurrently, timeout)
 import UnliftIO.Async (async)
+import UnliftIO.Concurrent (forkIO, mkWeakThreadId)
 import UnliftIO.Directory (doesFileExist, getTemporaryDirectory, removeFile)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -312,8 +313,7 @@ data AgentClient = AgentClient
     removedSubs :: TMap (UserId, SMPServer, SMP.RecipientId) SMPClientError,
     workerSeq :: TVar Int,
     smpDeliveryWorkers :: TMap SndQAddr (Worker, TMVar ()),
-    asyncCmdWorkers :: TMap (Maybe SMPServer) Worker,
-    connCmdsQueued :: TMap ConnId Bool,
+    asyncCmdWorkers :: TMap (ConnId, Maybe SMPServer) Worker,
     ntfNetworkOp :: TVar AgentOpState,
     rcvNetworkOp :: TVar AgentOpState,
     msgDeliveryOp :: TVar AgentOpState,
@@ -411,7 +411,7 @@ runWorkerAsync Worker {action} work =
     (atomically . tryPutTMVar action) -- if it was running (or if start crashes), put it back and unlock (don't lock if it was just started)
     (\a -> when (isNothing a) start) -- start worker if it's not running
   where
-    start = atomically . putTMVar action . Just =<< async work
+    start = atomically . putTMVar action . Just =<< mkWeakThreadId =<< forkIO work
 
 data AgentOperation = AONtfNetwork | AORcvNetwork | AOMsgDelivery | AOSndNetwork | AODatabase
   deriving (Eq, Show)
@@ -480,7 +480,6 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} currentTs a
   workerSeq <- newTVarIO 0
   smpDeliveryWorkers <- TM.emptyIO
   asyncCmdWorkers <- TM.emptyIO
-  connCmdsQueued <- TM.emptyIO
   ntfNetworkOp <- newTVarIO $ AgentOpState False 0
   rcvNetworkOp <- newTVarIO $ AgentOpState False 0
   msgDeliveryOp <- newTVarIO $ AgentOpState False 0
@@ -519,7 +518,6 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} currentTs a
         workerSeq,
         smpDeliveryWorkers,
         asyncCmdWorkers,
-        connCmdsQueued,
         ntfNetworkOp,
         rcvNetworkOp,
         msgDeliveryOp,
@@ -893,7 +891,6 @@ closeAgentClient c = do
   atomically (swapTVar (smpSubWorkers c) M.empty) >>= mapM_ cancelReconnect
   clearWorkers smpDeliveryWorkers >>= mapM_ (cancelWorker . fst)
   clearWorkers asyncCmdWorkers >>= mapM_ cancelWorker
-  clear connCmdsQueued
   atomically . RQ.clear $ activeSubs c
   atomically . RQ.clear $ pendingSubs c
   clear subscrConns
@@ -909,7 +906,7 @@ closeAgentClient c = do
 cancelWorker :: Worker -> IO ()
 cancelWorker Worker {doWork, action} = do
   noWorkToDo doWork
-  atomically (tryTakeTMVar action) >>= mapM_ (mapM_ uninterruptibleCancel)
+  atomically (tryTakeTMVar action) >>= mapM_ (mapM_ $ deRefWeak >=> mapM_ killThread)
 
 waitUntilActive :: AgentClient -> IO ()
 waitUntilActive AgentClient {active} = unlessM (readTVarIO active) $ atomically $ unlessM (readTVar active) retry

src/Simplex/Messaging/Agent/Env/SQLite.hs

@@ -41,6 +41,7 @@ module Simplex.Messaging.Agent.Env.SQLite
   )
 where
 
+import Control.Concurrent (ThreadId)
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
 import Control.Monad.Reader
@@ -76,8 +77,9 @@ import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (SMPVersion, TLS, Transport (..))
 import Simplex.Messaging.Transport.Client (defaultSMPPort)
 import Simplex.Messaging.Util (allFinally, catchAllErrors, catchAllErrors', tryAllErrors, tryAllErrors')
+import System.Mem.Weak (Weak)
 import System.Random (StdGen, newStdGen)
-import UnliftIO (Async, SomeException)
+import UnliftIO (SomeException)
 import UnliftIO.STM
 
 type AM' a = ReaderT Env IO a
@@ -312,7 +314,7 @@ mkInternal = INTERNAL . show
 data Worker = Worker
   { workerId :: Int,
     doWork :: TMVar (),
-    action :: TMVar (Maybe (Async ())),
+    action :: TMVar (Maybe (Weak ThreadId)),
     restarts :: TVar RestartCount
   }
 

src/Simplex/Messaging/Agent/Store/SQLite.hs

@@ -57,6 +57,7 @@ module Simplex.Messaging.Agent.Store.SQLite
     getDeletedConns,
     getConnData,
     setConnDeleted,
+    setConnUserId,
     setConnAgentVersion,
     setConnPQSupport,
     getDeletedConnIds,
@@ -110,6 +111,7 @@ module Simplex.Messaging.Agent.Store.SQLite
     getSndMsgViaRcpt,
     updateSndMsgRcpt,
     getPendingQueueMsg,
+    getConnectionsForDelivery,
     updatePendingMsgRIState,
     deletePendingMsgs,
     getExpiredSndMessages,
@@ -135,6 +137,7 @@ module Simplex.Messaging.Agent.Store.SQLite
     createCommand,
     getPendingCommandServers,
     getPendingServerCommand,
+    updateCommandServer,
     deleteCommand,
     -- Notification device token persistence
     createNtfToken,
@@ -1008,6 +1011,10 @@ updateSndMsgRcpt db connId sndMsgId MsgReceipt {agentMsgId, msgRcptStatus} =
     "UPDATE snd_messages SET rcpt_internal_id = ?, rcpt_status = ? WHERE conn_id = ? AND internal_snd_id = ?"
     (agentMsgId, msgRcptStatus, connId, sndMsgId)
 
+getConnectionsForDelivery :: DB.Connection -> IO [ConnId]
+getConnectionsForDelivery db =
+  map fromOnly <$> DB.query_ db "SELECT DISTINCT conn_id FROM snd_message_deliveries WHERE failed = 0"
+
 getPendingQueueMsg :: DB.Connection -> ConnId -> SndQueue -> IO (Either StoreError (Maybe (Maybe RcvQueue, PendingMsgData)))
 getPendingQueueMsg db connId SndQueue {dbQueueId} =
   getWorkItem "message" getMsgId getMsgData markMsgFailed
@@ -1317,38 +1324,39 @@ getPendingCommandServers db connId = do
   where
     smpServer (host, port, keyHash) = SMPServer <$> host <*> port <*> keyHash
 
-getPendingServerCommand :: DB.Connection -> Maybe SMPServer -> IO (Either StoreError (Maybe PendingCommand))
-getPendingServerCommand db srv_ = getWorkItem "command" getCmdId getCommand markCommandFailed
+getPendingServerCommand :: DB.Connection -> ConnId -> Maybe SMPServer -> IO (Either StoreError (Maybe PendingCommand))
+getPendingServerCommand db connId srv_ = getWorkItem "command" getCmdId getCommand markCommandFailed
   where
     getCmdId :: IO (Maybe Int64)
     getCmdId =
       maybeFirstRow fromOnly $ case srv_ of
         Nothing ->
-          DB.query_
+          DB.query
             db
             [sql|
               SELECT command_id FROM commands
-              WHERE host IS NULL AND port IS NULL AND failed = 0
+              WHERE conn_id = ? AND host IS NULL AND port IS NULL AND failed = 0
               ORDER BY created_at ASC, command_id ASC
               LIMIT 1
             |]
+            (Only connId)
         Just (SMPServer host port _) ->
           DB.query
             db
             [sql|
               SELECT command_id FROM commands
-              WHERE host = ? AND port = ? AND failed = 0
+              WHERE conn_id = ? AND host = ? AND port = ? AND failed = 0
               ORDER BY created_at ASC, command_id ASC
               LIMIT 1
             |]
-            (host, port)
+            (connId, host, port)
     getCommand :: Int64 -> IO (Either StoreError PendingCommand)
     getCommand cmdId =
       firstRow pendingCommand err $
         DB.query
           db
           [sql|
-            SELECT c.corr_id, cs.user_id, c.conn_id, c.command
+            SELECT c.corr_id, cs.user_id, c.command
             FROM commands c
             JOIN connections cs USING (conn_id)
             WHERE c.command_id = ?
@@ -1356,9 +1364,22 @@ getPendingServerCommand db srv_ = getWorkItem "command" getCmdId getCommand mark
           (Only cmdId)
       where
         err = SEInternal $ "command  " <> bshow cmdId <> " returned []"
-        pendingCommand (corrId, userId, connId, command) = PendingCommand {cmdId, corrId, userId, connId, command}
+        pendingCommand (corrId, userId, command) = PendingCommand {cmdId, corrId, userId, connId, command}
     markCommandFailed cmdId = DB.execute db "UPDATE commands SET failed = 1 WHERE command_id = ?" (Only cmdId)
 
+updateCommandServer :: DB.Connection -> AsyncCmdId -> SMPServer -> IO (Either StoreError ())
+updateCommandServer db cmdId srv@(SMPServer host port _) = runExceptT $ do
+  serverKeyHash_ <- ExceptT $ getServerKeyHash_ db srv
+  liftIO $
+    DB.execute
+      db
+      [sql|
+        UPDATE commands
+        SET host = ?, port = ?, server_key_hash = ?
+        WHERE command_id = ?
+      |]
+      (host, port, serverKeyHash_, cmdId)
+
 deleteCommand :: DB.Connection -> AsyncCmdId -> IO ()
 deleteCommand db cmdId =
   DB.execute db "DELETE FROM commands WHERE command_id = ?" (Only cmdId)
@@ -1909,9 +1930,11 @@ newQueueId_ (Only maxId : _) = DBQueueId (maxId + 1)
 
 getConn :: DB.Connection -> ConnId -> IO (Either StoreError SomeConn)
 getConn = getAnyConn False
+{-# INLINE getConn #-}
 
 getDeletedConn :: DB.Connection -> ConnId -> IO (Either StoreError SomeConn)
 getDeletedConn = getAnyConn True
+{-# INLINE getDeletedConn #-}
 
 getAnyConn :: Bool -> DB.Connection -> ConnId -> IO (Either StoreError SomeConn)
 getAnyConn deleted' dbConn connId =
@@ -1932,9 +1955,11 @@ getAnyConn deleted' dbConn connId =
 
 getConns :: DB.Connection -> [ConnId] -> IO [Either StoreError SomeConn]
 getConns = getAnyConns_ False
+{-# INLINE getConns #-}
 
 getDeletedConns :: DB.Connection -> [ConnId] -> IO [Either StoreError SomeConn]
 getDeletedConns = getAnyConns_ True
+{-# INLINE getDeletedConns #-}
 
 getAnyConns_ :: Bool -> DB.Connection -> [ConnId] -> IO [Either StoreError SomeConn]
 getAnyConns_ deleted' db connIds = forM connIds $ E.handle handleDBError . getAnyConn deleted' db
@@ -1967,6 +1992,10 @@ setConnDeleted db waitDelivery connId
   | otherwise =
       DB.execute db "UPDATE connections SET deleted = ? WHERE conn_id = ?" (True, connId)
 
+setConnUserId :: DB.Connection -> UserId -> ConnId -> UserId -> IO ()
+setConnUserId db oldUserId connId newUserId = 
+  DB.execute db "UPDATE connections SET user_id = ? WHERE conn_id = ? and user_id = ?" (newUserId, connId, oldUserId)
+
 setConnAgentVersion :: DB.Connection -> ConnId -> VersionSMPA -> IO ()
 setConnAgentVersion db connId aVersion =
   DB.execute db "UPDATE connections SET smp_agent_version = ? WHERE conn_id = ?" (aVersion, connId)

src/Simplex/Messaging/Agent/Store/SQLite/DB.hs

@@ -1,4 +1,5 @@
 {-# LANGUAGE DeriveAnyClass #-}
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE StrictData #-}
 {-# LANGUAGE TemplateHaskell #-}
@@ -20,15 +21,19 @@ where
 
 import Control.Concurrent.STM
 import Control.Monad (when)
+import Control.Exception
 import qualified Data.Aeson.TH as J
 import Data.Int (Int64)
+import Data.Map.Strict (Map)
+import qualified Data.Map.Strict as M
+import Data.Text (Text)
 import Data.Time (diffUTCTime, getCurrentTime)
 import Database.SQLite.Simple (FromRow, NamedParam, Query, ToRow)
 import qualified Database.SQLite.Simple as SQL
 import Simplex.Messaging.Parsers (defaultJSON)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (diffToMilliseconds)
+import Simplex.Messaging.Util (diffToMilliseconds, tshow)
 
 data Connection = Connection
   { conn :: SQL.Connection,
@@ -38,27 +43,35 @@ data Connection = Connection
 data SlowQueryStats = SlowQueryStats
   { count :: Int64,
     timeMax :: Int64,
-    timeAvg :: Int64
+    timeAvg :: Int64,
+    errs :: Map Text Int
   }
   deriving (Show)
 
 timeIt :: TMap Query SlowQueryStats -> Query -> IO a -> IO a
 timeIt slow sql a = do
   t <- getCurrentTime
-  r <- a
+  r <- a `catch` \e -> do
+    atomically $ TM.alter (Just . updateQueryErrors e) sql slow
+    throwIO e
   t' <- getCurrentTime
   let diff = diffToMilliseconds $ diffUTCTime t' t
-  atomically $ when (diff > 5) $ TM.alter (updateQueryStats diff) sql slow
+  when (diff > 1) $ atomically $ TM.alter (updateQueryStats diff) sql slow
   pure r
   where
+    updateQueryErrors :: SomeException -> Maybe SlowQueryStats -> SlowQueryStats
+    updateQueryErrors e Nothing = SlowQueryStats 0 0 0 $ M.singleton (tshow e) 1
+    updateQueryErrors e (Just stats@SlowQueryStats {errs}) =
+      stats {errs = M.alter (Just . maybe 1 (+ 1)) (tshow e) errs}
     updateQueryStats :: Int64 -> Maybe SlowQueryStats -> Maybe SlowQueryStats
-    updateQueryStats diff Nothing = Just $ SlowQueryStats 1 diff diff
-    updateQueryStats diff (Just SlowQueryStats {count, timeMax, timeAvg}) =
+    updateQueryStats diff Nothing = Just $ SlowQueryStats 1 diff diff M.empty
+    updateQueryStats diff (Just SlowQueryStats {count, timeMax, timeAvg, errs}) =
       Just $
         SlowQueryStats
           { count = count + 1,
             timeMax = max timeMax diff,
-            timeAvg = (timeAvg * count + diff) `div` (count + 1)
+            timeAvg = (timeAvg * count + diff) `div` (count + 1),
+            errs
           }
 
 open :: String -> IO Connection

src/Simplex/Messaging/Client/Agent.hs

@@ -372,14 +372,16 @@ smpSubscribeQueues party ca smp srv subs = do
     notify_ :: (SMPServer -> SMPSubParty -> NonEmpty a -> SMPClientAgentEvent) -> [a] -> IO ()
     notify_ evt qs = mapM_ (notify ca . evt srv party) $ L.nonEmpty qs
 
-activeClientSession :: SMPClientAgent -> SMPClient -> SMPServer -> STM Bool
-activeClientSession ca smp srv = sameSess <$> tryReadSessVar srv (smpClients ca)
+activeClientSession' :: SMPClientAgent -> SessionId -> SMPServer -> STM Bool
+activeClientSession' ca sessId srv = sameSess <$> tryReadSessVar srv (smpClients ca)
   where
-    sessId = sessionId . thParams
     sameSess = \case
-      Just (Right (_, smp')) -> sessId smp == sessId smp'
+      Just (Right (_, smp')) -> sessId == sessionId (thParams smp')
       _ -> False
 
+activeClientSession :: SMPClientAgent -> SMPClient -> SMPServer -> STM Bool
+activeClientSession ca = activeClientSession' ca . sessionId . thParams
+
 showServer :: SMPServer -> ByteString
 showServer ProtocolServer {host, port} =
   strEncode host <> B.pack (if null port then "" else ':' : port)

src/Simplex/Messaging/Notifications/Server.hs

@@ -203,7 +203,7 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
 
     receiveSMP :: M ()
     receiveSMP = forever $ do
-      ((_, srv, _), _, _, ts) <- atomically $ readTBQueue msgQ
+      ((_, srv, _), _thVersion, sessionId, ts) <- atomically $ readTBQueue msgQ
       forM ts $ \(ntfId, t) -> case t of
         STUnexpectedError e -> logError $ "SMP client unexpected error: " <> tshow e -- uncorrelated response, should not happen
         STResponse {} -> pure () -- it was already reported as timeout error
@@ -220,7 +220,9 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
                 findNtfSubscriptionToken st smpQueue
                   >>= mapM_ (\tkn -> writeTBQueue pushQ (tkn, PNMessage PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta}))
               incNtfStat ntfReceived
-            Right SMP.END -> updateSubStatus smpQueue NSEnd
+            Right SMP.END ->
+              whenM (atomically $ activeClientSession' ca sessionId srv) $
+                updateSubStatus smpQueue NSEnd
             Right (SMP.ERR e) -> logError $ "SMP server error: " <> tshow e
             Right _ -> logError "SMP server unexpected response"
             Left e -> logError $ "SMP client error: " <> tshow e

src/Simplex/Messaging/Notifications/Server/Main.hs

@@ -1,6 +1,7 @@
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedLists #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
@@ -132,7 +133,8 @@ ntfServerCLI cfgPath logPath =
                         { networkConfig =
                             defaultNetworkConfig
                               { socksProxy = either error id <$!> strDecodeIni "SUBSCRIBER" "socks_proxy" ini,
-                                socksMode = maybe SMOnion (either error id) $! strDecodeIni "SUBSCRIBER" "socks_mode" ini
+                                socksMode = maybe SMOnion (either error id) $! strDecodeIni "SUBSCRIBER" "socks_mode" ini,
+                                smpPingInterval = 60_000_000 -- 1 minutes
                               }
                         },
                     persistErrorInterval = 0 -- seconds

src/Simplex/Messaging/Server.hs

@@ -175,7 +175,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
           $>>= endPreviousSubscriptions
           >>= liftIO . mapM_ unsub
       where
-        updateSubscribers :: TVar (IM.IntMap Client) -> STM (Maybe (QueueId, Client))
+        updateSubscribers :: TVar (IM.IntMap (Maybe Client)) -> STM (Maybe (QueueId, Client))
         updateSubscribers cls = do
           (qId, clnt, subscribed) <- readTQueue $ subQ s
           current <- IM.member (clientId clnt) <$> readTVar cls
@@ -412,7 +412,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
               CPClients -> withAdminRole $ do
                 active <- unliftIO u (asks clients) >>= readTVarIO
                 hPutStrLn h "clientId,sessionId,connected,createdAt,rcvActiveAt,sndActiveAt,age,subscriptions"
-                forM_ (IM.toList active) $ \(cid, Client {sessionId, connected, createdAt, rcvActiveAt, sndActiveAt, subscriptions}) -> do
+                forM_ (IM.toList active) $ \(cid, cl) -> forM_ cl $ \Client {sessionId, connected, createdAt, rcvActiveAt, sndActiveAt, subscriptions} -> do
                   connected' <- bshow <$> readTVarIO connected
                   rcvActiveAt' <- strEncode <$> readTVarIO rcvActiveAt
                   sndActiveAt' <- strEncode <$> readTVarIO sndActiveAt
@@ -507,7 +507,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                   activeClients <- readTVarIO clients
                   hPutStrLn h $ "Clients: " <> show (IM.size activeClients)
                   when (r == CPRAdmin) $ do
-                    clQs <- clientTBQueueLengths activeClients
+                    clQs <- clientTBQueueLengths' activeClients
                     hPutStrLn h $ "Client queues (rcvQ, sndQ, msgQ): " <> show clQs
                     (smpSubCnt, smpSubCntByGroup, smpClCnt, smpClQs) <- countClientSubs subscriptions (Just countSMPSubs) activeClients
                     hPutStrLn h $ "SMP subscriptions (via clients): " <> show smpSubCnt
@@ -542,11 +542,12 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                               | otherwise = (cl : cls, IS.insert clientId clSet)
                         countSubClients :: M.Map QueueId Client -> Int
                         countSubClients = IS.size . M.foldr' (IS.insert . clientId) IS.empty
-                    countClientSubs :: (Client -> TMap QueueId a) -> Maybe (M.Map QueueId a -> IO (Int, Int, Int, Int)) -> IM.IntMap Client -> IO (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural))
+                    countClientSubs :: (Client -> TMap QueueId a) -> Maybe (M.Map QueueId a -> IO (Int, Int, Int, Int)) -> IM.IntMap (Maybe Client) -> IO (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural))
                     countClientSubs subSel countSubs_ = foldM addSubs (0, (0, 0, 0, 0), 0, (0, 0, 0))
                       where
-                        addSubs :: (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural)) -> Client -> IO (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural))
-                        addSubs (!subCnt, cnts@(!c1, !c2, !c3, !c4), !clCnt, !qs) cl = do
+                        addSubs :: (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural)) -> Maybe Client -> IO (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural))
+                        addSubs acc Nothing = pure acc
+                        addSubs (!subCnt, cnts@(!c1, !c2, !c3, !c4), !clCnt, !qs) (Just cl) = do
                           subs <- readTVarIO $ subSel cl
                           cnts' <- case countSubs_ of
                             Nothing -> pure cnts
@@ -559,6 +560,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                           pure (subCnt + cnt, cnts', clCnt', qs')
                     clientTBQueueLengths :: Foldable t => t Client -> IO (Natural, Natural, Natural)
                     clientTBQueueLengths = foldM addQueueLengths (0, 0, 0)
+                    clientTBQueueLengths' :: Foldable t => t (Maybe Client) -> IO (Natural, Natural, Natural)
+                    clientTBQueueLengths' = foldM (\acc -> maybe (pure acc) (addQueueLengths acc)) (0, 0, 0)
                     addQueueLengths (!rl, !sl, !ml) cl = do
                       (rl', sl', ml') <- queueLengths cl
                       pure (rl + rl', sl + sl', ml + ml')
@@ -619,15 +622,18 @@ runClientTransport h@THandle {params = thParams@THandleParams {thVersion, sessio
   ts <- liftIO getSystemTime
   active <- asks clients
   nextClientId <- asks clientSeq
-  c@Client {clientId} <- liftIO $ newClient nextClientId q thVersion sessionId ts
-  atomically $ modifyTVar' active $ IM.insert clientId c
-  s <- asks server
-  expCfg <- asks $ inactiveClientExpiration . config
-  th <- newMVar h -- put TH under a fair lock to interleave messages and command responses
-  labelMyThread . B.unpack $ "client $" <> encode sessionId
-  raceAny_ ([liftIO $ send th c, liftIO $ sendMsg th c, client thParams c s, receive h c] <> disconnectThread_ c expCfg)
-    `finally` clientDisconnected c
+  clientId <- atomically $ stateTVar nextClientId $ \next -> (next, next + 1)
+  atomically $ modifyTVar' active $ IM.insert clientId Nothing
+  c <- liftIO $ newClient clientId q thVersion sessionId ts
+  runClientThreads active c clientId `finally` clientDisconnected c
   where
+    runClientThreads active c clientId = do
+      atomically $ modifyTVar' active $ IM.insert clientId $ Just c
+      s <- asks server
+      expCfg <- asks $ inactiveClientExpiration . config
+      th <- newMVar h -- put TH under a fair lock to interleave messages and command responses
+      labelMyThread . B.unpack $ "client $" <> encode sessionId
+      raceAny_ $ [liftIO $ send th c, liftIO $ sendMsg th c, client thParams c s, receive h c] <> disconnectThread_ c expCfg
     disconnectThread_ c (Just expCfg) = [liftIO $ disconnectTransport h (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c)]
     disconnectThread_ _ _ = []
     noSubscriptions c = atomically $ (&&) <$> TM.null (ntfSubscriptions c) <*> (not . hasSubs <$> readTVar (subscriptions c))

src/Simplex/Messaging/Server/Env/STM.hs

@@ -127,7 +127,7 @@ data Env = Env
     serverStats :: ServerStats,
     sockets :: SocketState,
     clientSeq :: TVar ClientId,
-    clients :: TVar (IntMap Client),
+    clients :: TVar (IntMap (Maybe Client)),
     proxyAgent :: ProxyAgent -- senders served on this proxy
   }
 
@@ -183,9 +183,8 @@ newServer = do
   savingLock <- atomically createLock
   return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers, savingLock}
 
-newClient :: TVar ClientId -> Natural -> VersionSMP -> ByteString -> SystemTime -> IO Client
-newClient nextClientId qSize thVersion sessionId createdAt = do
-  clientId <- atomically $ stateTVar nextClientId $ \next -> (next, next + 1)
+newClient :: ClientId -> Natural -> VersionSMP -> ByteString -> SystemTime -> IO Client
+newClient clientId qSize thVersion sessionId createdAt = do
   subscriptions <- TM.emptyIO
   ntfSubscriptions <- TM.emptyIO
   rcvQ <- newTBQueueIO qSize

src/Simplex/Messaging/Server/StoreLog.hs

@@ -36,7 +36,7 @@ import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.QueueStore (NtfCreds (..), QueueRec (..), ServerQueueStatus (..))
 import Simplex.Messaging.Transport.Buffer (trimCR)
 import Simplex.Messaging.Util (ifM)
-import System.Directory (doesFileExist)
+import System.Directory (doesFileExist, renameFile)
 import System.IO
 
 -- | opaque container for file handle with a type-safe IOMode
@@ -140,10 +140,12 @@ logDeleteNotifier s = writeStoreLogRecord s . DeleteNotifier
 
 readWriteStoreLog :: FilePath -> IO (Map RecipientId QueueRec, StoreLog 'WriteMode)
 readWriteStoreLog f = do
-  qs <- ifM (doesFileExist f) (readQueues f) (pure M.empty)
+  qs <- ifM (doesFileExist f) readQS (pure M.empty)
   s <- openWriteStoreLog f
   writeQueues s qs
   pure (qs, s)
+  where
+    readQS = readQueues f <* renameFile f (f <> ".bak")
 
 writeQueues :: StoreLog 'WriteMode -> Map RecipientId QueueRec -> IO ()
 writeQueues s = mapM_ $ \q -> when (active q) $ logCreateQueue s q

src/Simplex/Messaging/Transport.hs

@@ -113,6 +113,7 @@ import Simplex.Messaging.Transport.Buffer
 import Simplex.Messaging.Util (bshow, catchAll, catchAll_, liftEitherWith)
 import Simplex.Messaging.Version
 import Simplex.Messaging.Version.Internal
+import System.IO.Error (isEOFError)
 import UnliftIO.Exception (Exception)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -339,11 +340,12 @@ instance Transport TLS where
 
   getLn :: TLS -> IO ByteString
   getLn TLS {tlsContext, tlsBuffer} = do
-    getLnBuffered tlsBuffer (T.recvData tlsContext) `E.catch` handleEOF
+    getLnBuffered tlsBuffer (T.recvData tlsContext) `E.catches` [E.Handler handleTlsEOF, E.Handler handleEOF]
     where
-      handleEOF = \case
-        T.Error_EOF -> E.throwIO TEBadBlock
+      handleTlsEOF = \case
+        T.PostHandshake T.Error_EOF -> E.throwIO TEBadBlock
         e -> E.throwIO e
+      handleEOF e = if isEOFError e then E.throwIO TEBadBlock else E.throwIO e
 
 -- * SMP transport
 

src/Simplex/Messaging/Transport/WebSockets.hs

@@ -25,6 +25,7 @@ import Simplex.Messaging.Transport
     withTlsUnique,
   )
 import Simplex.Messaging.Transport.Buffer (trimCR)
+import System.IO.Error (isEOFError)
 
 data WS = WS
   { wsPeer :: TransportPeer,
@@ -108,9 +109,11 @@ makeTLSContextStream cxt =
   S.makeStream readStream writeStream
   where
     readStream :: IO (Maybe ByteString)
-    readStream =
-      (Just <$> T.recvData cxt) `E.catch` \case
-        T.Error_EOF -> pure Nothing
-        e -> E.throwIO e
+    readStream = (Just <$> T.recvData cxt) `E.catches` [E.Handler handleTlsEOF, E.Handler handleEOF]
+      where
+        handleTlsEOF = \case
+          T.PostHandshake T.Error_EOF -> pure Nothing
+          e -> E.throwIO e
+        handleEOF e = if isEOFError e then pure Nothing else E.throwIO e
     writeStream :: Maybe LB.ByteString -> IO ()
     writeStream = maybe (closeTLS cxt) (T.sendData cxt)

src/Simplex/Messaging/Util.hs

@@ -166,10 +166,10 @@ threadDelay' = loop
           loop $ time - maxWait
 
 diffToMicroseconds :: NominalDiffTime -> Int64
-diffToMicroseconds diff = fromIntegral ((truncate $ diff * 1000000) :: Integer)
+diffToMicroseconds diff = truncate $ diff * 1000000
 
 diffToMilliseconds :: NominalDiffTime -> Int64
-diffToMilliseconds diff = fromIntegral ((truncate $ diff * 1000) :: Integer)
+diffToMilliseconds diff = truncate $ diff * 1000
 
 labelMyThread :: MonadIO m => String -> m ()
 labelMyThread label = liftIO $ myThreadId >>= (`labelThread` label)

src/Simplex/RemoteControl/Client.hs

@@ -305,7 +305,7 @@ connectRCCtrl_ drg pairing'@RCCtrlPairing {caKey, caCert} inv@RCInvitation {ca,
 
 catchRCError :: ExceptT RCErrorType IO a -> (RCErrorType -> ExceptT RCErrorType IO a) -> ExceptT RCErrorType IO a
 catchRCError = catchAllErrors $ \e -> case fromException e of
-  Just (TLS.Terminated _ _ (TLS.Error_Protocol (_, _, TLS.UnknownCa))) -> RCEIdentity
+  Just (TLS.Terminated _ _ (TLS.Error_Protocol _ TLS.UnknownCa)) -> RCEIdentity
   _ -> RCEException $ show e
 {-# INLINE catchRCError #-}
 

tests/AgentTests/FunctionalAPITests.hs

@@ -283,6 +283,9 @@ functionalAPITests t = do
     testPQMatrix3 t $ runAgentClientContactTestPQ3 True
   it "should support rejecting contact request" $
     withSmpServer t testRejectContactRequest
+  describe "Changing connection user id" $ do
+    it "should change user id for new connections" $ do
+      withSmpServer t testUpdateConnectionUserId
   describe "Establishing connection asynchronously" $ do
     it "should connect with initiating client going offline" $
       withSmpServer t testAsyncInitiatingOffline
@@ -912,6 +915,25 @@ testRejectContactRequest =
     rejectContact alice addrConnId invId
     liftIO $ noMessages bob "nothing delivered to bob"
 
+testUpdateConnectionUserId :: HasCallStack => IO ()
+testUpdateConnectionUserId =
+  withAgentClients2 $ \alice bob -> runRight_ $ do
+    (connId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
+    newUserId <- createUser alice [noAuthSrvCfg testSMPServer] [noAuthSrvCfg testXFTPServer]
+    _ <- changeConnectionUser alice 1 connId newUserId
+    aliceId <- A.prepareConnectionToJoin bob 1 True qInfo PQSupportOn
+    (aliceId', sqSecured') <- A.joinConnection bob 1 (Just aliceId) True qInfo "bob's connInfo" PQSupportOn SMSubscribe
+    liftIO $ do
+      aliceId' `shouldBe` aliceId
+      sqSecured' `shouldBe` True
+    ("", _, A.CONF confId pqSup' _ "bob's connInfo") <- get alice
+    liftIO $ pqSup' `shouldBe` PQSupportOn
+    allowConnection alice connId confId "alice's connInfo"
+    let pqEnc = CR.pqSupportToEnc PQSupportOn
+    get alice ##> ("", connId, A.CON pqEnc)
+    get bob ##> ("", aliceId, A.INFO PQSupportOn "alice's connInfo")
+    get bob ##> ("", aliceId, A.CON pqEnc)
+
 testAsyncInitiatingOffline :: HasCallStack => IO ()
 testAsyncInitiatingOffline =
   withAgent 2 agentCfg initAgentServers testDB2 $ \bob -> runRight_ $ do
@@ -1041,13 +1063,12 @@ testAllowConnectionClientRestart t = do
     threadDelay 250000
 
     alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
+    runRight_ $ subscribeConnection alice2 bobId
+    threadDelay 500000
 
     withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do
       runRight $ do
         ("", "", UP _ _) <- nGet bob
-
-        subscribeConnection alice2 bobId
-
         get alice2 ##> ("", bobId, CON)
         get bob ##> ("", aliceId, INFO "alice's connInfo")
         get bob ##> ("", aliceId, CON)

tests/AgentTests/NotificationTests.hs

@@ -508,6 +508,7 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} baseId ali
   suspendAgent alice 0
   closeSQLiteStore store
   threadDelay 1000000
+  putStrLn "before opening the database from another agent"
 
   -- aliceNtf client doesn't have subscription and is allowed to get notification message
   withAgent 3 aliceCfg initAgentServers testDB $ \aliceNtf -> runRight_ $ do
@@ -515,6 +516,7 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} baseId ali
     pure ()
 
   threadDelay 1000000
+  putStrLn "after closing the database in another agent"
   reopenSQLiteStore store
   foregroundAgent alice
   threadDelay 500000

tests/AgentTests/SQLiteTests.hs

@@ -114,6 +114,8 @@ storeTests = do
           testDeleteRcvConn
           testDeleteSndConn
           testDeleteDuplexConn
+        describe "setConnUserId" $ do
+          testSetConnUserIdNewConn
         describe "upgradeRcvConnToDuplex" $ do
           testUpgradeRcvConnToDuplex
         describe "upgradeSndConnToDuplex" $ do
@@ -336,6 +338,21 @@ testGetRcvConn =
     getRcvConn db smpServer recipientId
       `shouldReturn` Right (rq, SomeConn SCRcv (RcvConnection cData1 rq))
 
+testSetConnUserIdNewConn :: SpecWith SQLiteStore
+testSetConnUserIdNewConn =
+  it "should set user id for new connection" . withStoreTransaction $ \db -> do
+    g <- C.newRandom
+    Right connId <- createNewConn db g cData1 {connId = ""} SCMInvitation
+    newUserId <- createUserRecord db
+    _ <- setConnUserId db 1 connId newUserId
+    connResult <- getConn db connId
+    case connResult of
+      Right (SomeConn SCNew (NewConnection connData)) -> do
+        let ConnData {userId} = connData
+        userId `shouldBe` newUserId
+      _ -> do
+         expectationFailure "Failed to get connection"
+
 testDeleteRcvConn :: SpecWith SQLiteStore
 testDeleteRcvConn =
   it "should create RcvConnection and delete it" . withStoreTransaction $ \db -> do
@@ -644,30 +661,30 @@ testGetPendingServerCommand :: SQLiteStore -> Expectation
 testGetPendingServerCommand st = do
   g <- C.newRandom
   withTransaction st $ \db -> do
-    Right Nothing <- getPendingServerCommand db Nothing
+    Right Nothing <- getPendingServerCommand db "" Nothing
     Right connId <- createNewConn db g cData1 {connId = ""} SCMInvitation
     Right () <- createCommand db "1" connId Nothing command
     corruptCmd db "1" connId
     Right () <- createCommand db "2" connId Nothing command
 
-    Left e <- getPendingServerCommand db Nothing
+    Left e <- getPendingServerCommand db connId Nothing
     show e `shouldContain` "bad AgentCmdType"
     DB.query_ db "SELECT conn_id, corr_id FROM commands WHERE failed = 1" `shouldReturn` [(connId, "1" :: ByteString)]
 
-    Right (Just PendingCommand {corrId}) <- getPendingServerCommand db Nothing
+    Right (Just PendingCommand {corrId}) <- getPendingServerCommand db connId Nothing
     corrId `shouldBe` "2"
 
     Right _ <- updateNewConnRcv db connId rcvQueue1
-    Right Nothing <- getPendingServerCommand db $ Just smpServer1
+    Right Nothing <- getPendingServerCommand db connId $ Just smpServer1
     Right () <- createCommand db "3" connId (Just smpServer1) command
     corruptCmd db "3" connId
     Right () <- createCommand db "4" connId (Just smpServer1) command
 
-    Left e' <- getPendingServerCommand db (Just smpServer1)
+    Left e' <- getPendingServerCommand db connId (Just smpServer1)
     show e' `shouldContain` "bad AgentCmdType"
     DB.query_ db "SELECT conn_id, corr_id FROM commands WHERE failed = 1" `shouldReturn` [(connId, "1" :: ByteString), (connId, "3" :: ByteString)]
 
-    Right (Just PendingCommand {corrId = corrId'}) <- getPendingServerCommand db (Just smpServer1)
+    Right (Just PendingCommand {corrId = corrId'}) <- getPendingServerCommand db connId (Just smpServer1)
     corrId' `shouldBe` "4"
   where
     command = AClientCommand $ NEW True (ACM SCMInvitation) (IKNoPQ PQSupportOn) SMSubscribe