further analysis

Stale postgres schema leaked pending XFTP operations between
cross-language tests, causing N-1 of N tests to fail.

bench/ClientSim.hs

@@ -1,133 +0,0 @@
-{-# LANGUAGE DataKinds #-}
-{-# LANGUAGE DuplicateRecordFields #-}
-{-# LANGUAGE GADTs #-}
-{-# LANGUAGE LambdaCase #-}
-{-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE PatternSynonyms #-}
-{-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE TupleSections #-}
-{-# LANGUAGE TypeApplications #-}
-{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
-
-module ClientSim
-  ( SimClient (..),
-    connectClient,
-    createQueue,
-    subscribeQueue,
-    sendMessage,
-    receiveAndAck,
-    connectN,
-    benchKeyHash,
-  )
-where
-
-import Control.Concurrent.Async (mapConcurrently)
-import Control.Concurrent.STM
-import Control.Monad (forM_)
-import Control.Monad.Except (runExceptT)
-import Data.ByteString.Char8 (ByteString)
-import Data.List (unfoldr)
-import qualified Data.List.NonEmpty as L
-import Network.Socket (ServiceName)
-import qualified Simplex.Messaging.Crypto as C
-import Simplex.Messaging.Encoding
-import Simplex.Messaging.Protocol
-import Simplex.Messaging.Transport
-import Simplex.Messaging.Transport.Client
-import Simplex.Messaging.Version
-
-data SimClient = SimClient
-  { scHandle :: THandleSMP TLS 'TClient,
-    scRcvKey :: C.APrivateAuthKey,
-    scRcvId :: RecipientId,
-    scSndId :: SenderId,
-    scDhSecret :: C.DhSecret 'C.X25519
-  }
-
-benchKeyHash :: C.KeyHash
-benchKeyHash = "LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI="
-
-connectClient :: TransportHost -> ServiceName -> IO (THandleSMP TLS 'TClient)
-connectClient host port = do
-  let tcConfig = defaultTransportClientConfig {clientALPN = Just alpnSupportedSMPHandshakes}
-  runTransportClient tcConfig Nothing host port (Just benchKeyHash) $ \h ->
-    runExceptT (smpClientHandshake h Nothing benchKeyHash supportedClientSMPRelayVRange False Nothing) >>= \case
-      Right th -> pure th
-      Left e -> error $ "SMP handshake failed: " <> show e
-
-connectN :: Int -> TransportHost -> ServiceName -> IO [THandleSMP TLS 'TClient]
-connectN n host port = do
-  let batches = chunksOf 100 [1 .. n]
-  concat <$> mapM (\batch -> mapConcurrently (\_ -> connectClient host port) batch) batches
-
-createQueue :: THandleSMP TLS 'TClient -> IO SimClient
-createQueue h = do
-  g <- C.newRandom
-  (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
-  (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-  (dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
-  -- NEW command
-  Resp "1" NoEntity (Ids rId sId srvDh) <- signSendRecv h rKey ("1", NoEntity, New rPub dhPub)
-  let dhShared = C.dh' srvDh dhPriv
-  -- KEY command (secure queue)
-  Resp "2" _ OK <- signSendRecv h rKey ("2", rId, KEY sPub)
-  pure SimClient {scHandle = h, scRcvKey = rKey, scRcvId = rId, scSndId = sId, scDhSecret = dhShared}
-
-subscribeQueue :: SimClient -> IO ()
-subscribeQueue SimClient {scHandle = h, scRcvKey = rKey, scRcvId = rId} = do
-  Resp "3" _ (SOK _) <- signSendRecv h rKey ("3", rId, SUB)
-  pure ()
-
-sendMessage :: THandleSMP TLS 'TClient -> C.APrivateAuthKey -> SenderId -> ByteString -> IO ()
-sendMessage h sKey sId body = do
-  Resp "4" _ OK <- signSendRecv h sKey ("4", sId, SEND noMsgFlags body)
-  pure ()
-
-receiveAndAck :: SimClient -> IO ()
-receiveAndAck SimClient {scHandle = h, scRcvKey = rKey, scRcvId = rId} = do
-  (_, _, Right (MSG RcvMessage {msgId = mId})) <- tGet1 h
-  Resp "5" _ OK <- signSendRecv h rKey ("5", rId, ACK mId)
-  pure ()
-
--- Helpers (same patterns as ServerTests.hs)
-
-pattern Resp :: CorrId -> EntityId -> BrokerMsg -> Transmission (Either ErrorType BrokerMsg)
-pattern Resp corrId queueId command <- (corrId, queueId, Right command)
-
-pattern Ids :: RecipientId -> SenderId -> RcvPublicDhKey -> BrokerMsg
-pattern Ids rId sId srvDh <- IDS (QIK rId sId srvDh _ _ Nothing Nothing)
-
-pattern New :: RcvPublicAuthKey -> RcvPublicDhKey -> Command 'Creator
-pattern New rPub dhPub = NEW (NewQueueReq rPub dhPub Nothing SMSubscribe (Just (QRMessaging Nothing)) Nothing)
-
-signSendRecv :: (Transport c, PartyI p) => THandleSMP c 'TClient -> C.APrivateAuthKey -> (ByteString, EntityId, Command p) -> IO (Transmission (Either ErrorType BrokerMsg))
-signSendRecv h pk t = do
-  signSend h pk t
-  (r L.:| _) <- tGetClient h
-  pure r
-
-signSend :: (Transport c, PartyI p) => THandleSMP c 'TClient -> C.APrivateAuthKey -> (ByteString, EntityId, Command p) -> IO ()
-signSend h@THandle {params} (C.APrivateAuthKey a pk) (corrId, qId, cmd) = do
-  let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth params (CorrId corrId, qId, cmd)
-      authorize t = (,Nothing) <$> case a of
-        C.SEd25519 -> Just . TASignature . C.ASignature C.SEd25519 $ C.sign' pk t
-        C.SEd448 -> Just . TASignature . C.ASignature C.SEd448 $ C.sign' pk t
-        C.SX25519 -> (\THAuthClient {peerServerPubKey = k} -> TAAuthenticator $ C.cbAuthenticate k pk (C.cbNonce corrId) t) <$> thAuth params
-  Right () <- tPut1 h (authorize tForAuth, tToSend)
-  pure ()
-
-tPut1 :: Transport c => THandle v c 'TClient -> SentRawTransmission -> IO (Either TransportError ())
-tPut1 h t = do
-  rs <- tPut h (Right t L.:| [])
-  case rs of
-    (r : _) -> pure r
-    [] -> error "tPut1: empty result"
-
-tGet1 :: (ProtocolEncoding v err cmd, Transport c) => THandle v c 'TClient -> IO (Transmission (Either err cmd))
-tGet1 h = do
-  (r L.:| _) <- tGetClient h
-  pure r
-
-chunksOf :: Int -> [a] -> [[a]]
-chunksOf n = unfoldr $ \xs -> if null xs then Nothing else Just (splitAt n xs)

bench/Dockerfile

@@ -1,25 +0,0 @@
-FROM haskell:9.6.3 AS build
-
-WORKDIR /src
-
-# Copy cabal file first for dependency caching
-COPY simplexmq.cabal cabal.project* ./
-RUN cabal update && cabal build --only-dependencies -f server_postgres smp-server-bench || true
-
-# Copy full source
-COPY . .
-RUN cabal build -f server_postgres smp-server-bench \
-  && cp $(cabal list-bin -f server_postgres smp-server-bench) /usr/local/bin/smp-server-bench
-
-FROM debian:bookworm-slim
-
-RUN apt-get update && apt-get install -y --no-install-recommends \
-      libgmp10 libpq5 libffi8 zlib1g ca-certificates \
-  && rm -rf /var/lib/apt/lists/*
-
-COPY --from=build /usr/local/bin/smp-server-bench /usr/local/bin/smp-server-bench
-COPY tests/fixtures /app/tests/fixtures
-
-WORKDIR /app
-
-ENTRYPOINT ["smp-server-bench"]

bench/Main.hs

@@ -1,243 +0,0 @@
-{-# LANGUAGE CPP #-}
-{-# LANGUAGE DataKinds #-}
-{-# LANGUAGE DuplicateRecordFields #-}
-{-# LANGUAGE GADTs #-}
-{-# LANGUAGE LambdaCase #-}
-{-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE NumericUnderscores #-}
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE TypeApplications #-}
-{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
-
-module Main where
-
-import Control.Concurrent (threadDelay)
-import Control.Concurrent.Async (async, cancel, forConcurrently_, mapConcurrently, mapConcurrently_)
-import Control.Concurrent.STM
-import Control.Monad (forever, forM_, void, when)
-import Data.ByteString.Char8 (ByteString)
-import qualified Data.ByteString.Char8 as B
-import Data.IORef
-import Data.List (unfoldr)
-import Data.Time.Clock (getCurrentTime, utctDayTime)
-import Network.Socket (ServiceName)
-import System.Environment (getArgs)
-import System.IO (hFlush, stdout)
-
-import ClientSim
-import Report
-
-import Crypto.Random (ChaChaDRG)
-import qualified Simplex.Messaging.Crypto as C
-import Simplex.Messaging.Server (runSMPServerBlocking)
-import Simplex.Messaging.Server.Env.STM as Env
-import Simplex.Messaging.Server.Expiration (ExpirationConfig (..))
-import Simplex.Messaging.Server.MsgStore.Postgres (PostgresMsgStore)
-import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
-import Simplex.Messaging.Agent.Store.Postgres.Options (DBOpts (..))
-import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..))
-import Simplex.Messaging.Client.Agent (SMPClientAgentConfig (..), defaultSMPClientAgentConfig)
-import Simplex.Messaging.Transport
-import Simplex.Messaging.Transport.Server (ServerCredentials (..), mkTransportServerConfig)
-import Simplex.Messaging.Version
-import UnliftIO.Exception (bracket)
-
-import Control.Logger.Simple (logInfo, withGlobalLogging, LogConfig (..), setLogLevel, LogLevel (..))
-
-data BenchConfig = BenchConfig
-  { numClients :: Int,
-    sustainedMinutes :: Int,
-    pgConnStr :: ByteString,
-    serverPort :: ServiceName,
-    timeSeriesFile :: FilePath
-  }
-
-defaultBenchConfig :: BenchConfig
-defaultBenchConfig =
-  BenchConfig
-    { numClients = 5000,
-      sustainedMinutes = 5,
-      pgConnStr = "postgresql://smp@localhost:15432/smp_bench",
-      serverPort = "15001",
-      timeSeriesFile = "bench-timeseries.csv"
-    }
-
-parseArgs :: IO BenchConfig
-parseArgs = do
-  args <- getArgs
-  pure $ go args defaultBenchConfig
-  where
-    go [] c = c
-    go ("--clients" : n : rest) c = go rest c {numClients = read n}
-    go ("--minutes" : n : rest) c = go rest c {sustainedMinutes = read n}
-    go ("--pg" : s : rest) c = go rest c {pgConnStr = B.pack s}
-    go ("--port" : p : rest) c = go rest c {serverPort = p}
-    go ("--timeseries" : f : rest) c = go rest c {timeSeriesFile = f}
-    go (x : _) _ = error $ "Unknown argument: " <> x
-
-main :: IO ()
-main = withGlobalLogging LogConfig {lc_file = Nothing, lc_stderr = True} $ do
-  setLogLevel LogInfo
-  bc@BenchConfig {numClients, sustainedMinutes, serverPort, timeSeriesFile, pgConnStr} <- parseArgs
-  putStrLn $ "SMP Server Memory Benchmark"
-  putStrLn $ "  clients:  " <> show numClients
-  putStrLn $ "  sustain:  " <> show sustainedMinutes <> " min"
-  putStrLn $ "  pg:       " <> B.unpack pgConnStr
-  putStrLn $ "  port:     " <> serverPort
-  putStrLn ""
-
-  snapshotsRef <- newIORef []
-
-  let snap phase clients = do
-        s <- takeSnapshot phase clients
-        modifyIORef' snapshotsRef (s :)
-        putStrLn $ "  [" <> show phase <> "] live=" <> show (snapLive s `div` (1024 * 1024)) <> "MB large=" <> show (snapLarge s `div` (1024 * 1024)) <> "MB"
-        hFlush stdout
-
-  withBenchServer bc $ do
-    putStrLn "Phase 1: Baseline (no clients)"
-    snap "baseline" 0
-
-    putStrLn $ "Phase 2: Connecting " <> show numClients <> " TLS clients..."
-    handles <- connectN numClients "localhost" serverPort
-    putStrLn $ "  Connected " <> show (length handles) <> " clients"
-    snap "tls_connect" (length handles)
-
-    putStrLn "Phase 3: Creating queues (NEW + KEY)..."
-    simClients <- mapConcurrently createQueue handles
-    putStrLn $ "  Created " <> show (length simClients) <> " queues"
-    snap "queue_create" (length simClients)
-
-    putStrLn "Phase 4: Subscribing (SUB)..."
-    mapConcurrently_ subscribeQueue simClients
-    snap "subscribe" (length simClients)
-
-    -- Pair up clients: first half sends to second half
-    let halfN = length simClients `div` 2
-        senders = take halfN simClients
-        receivers = drop halfN simClients
-        pairs = zip senders receivers
-
-    putStrLn $ "Phase 5: Sending " <> show halfN <> " messages..."
-    g <- C.newRandom
-    forConcurrently_ pairs $ \(sender, receiver) -> do
-      (_, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-      sendMessage (scHandle sender) sKey (scSndId receiver) "benchmark test message payload 1234567890"
-    snap "msg_send" (length simClients)
-
-    putStrLn "Phase 6: Receiving and ACKing messages..."
-    forConcurrently_ receivers receiveAndAck
-    snap "msg_recv" (length simClients)
-
-    putStrLn $ "Phase 7: Sustained load (" <> show sustainedMinutes <> " min)..."
-    writeTimeSeriesHeader timeSeriesFile
-    -- Logger thread: snapshot every 10s
-    logger <- async $ forever $ do
-      threadDelay 10_000_000
-      s <- takeSnapshot "sustained" (length simClients)
-      appendTimeSeries timeSeriesFile s
-    -- Worker threads: continuous send/receive
-    let loopDurationUs = sustainedMinutes * 60 * 1_000_000
-    workersDone <- newTVarIO False
-    workers <- async $ do
-      deadline <- (+ loopDurationUs) <$> getMonotonicTimeUs
-      sustainedLoop g pairs deadline
-      atomically $ writeTVar workersDone True
-    -- Wait for workers
-    void $ atomically $ readTVar workersDone >>= \done -> when (not done) retry
-    cancel logger
-    cancel workers
-    snap "sustained_end" (length simClients)
-
-  snapshots <- reverse <$> readIORef snapshotsRef
-  printSummary snapshots
-  putStrLn $ "\nTime-series written to: " <> timeSeriesFile
-
-sustainedLoop :: TVar ChaChaDRG -> [(SimClient, SimClient)] -> Int -> IO ()
-sustainedLoop g pairs deadline = go
-  where
-    go = do
-      now <- getMonotonicTimeUs
-      when (now < deadline) $ do
-        forConcurrently_ pairs $ \(sender, receiver) -> do
-          (_, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-          sendMessage (scHandle sender) sKey (scSndId receiver) "sustained load message payload"
-        forConcurrently_ (map snd pairs) receiveAndAck
-        go
-
-getMonotonicTimeUs :: IO Int
-getMonotonicTimeUs = do
-  t <- getCurrentTime
-  pure $ round (utctDayTime t * 1_000_000)
-
-withBenchServer :: BenchConfig -> IO a -> IO a
-withBenchServer BenchConfig {pgConnStr, serverPort} action = do
-  started <- newEmptyTMVarIO
-  let srvCfg = benchServerConfig pgConnStr serverPort
-  bracket
-    (async $ runSMPServerBlocking started srvCfg Nothing)
-    cancel
-    (\_ -> waitForServer started >> action)
-  where
-    waitForServer started = do
-      r <- atomically $ takeTMVar started
-      if r
-        then putStrLn $ "Server started on port " <> serverPort
-        else error "Server failed to start"
-
-benchServerConfig :: ByteString -> ServiceName -> ServerConfig PostgresMsgStore
-benchServerConfig pgConn port =
-  let storeCfg = PostgresStoreCfg
-        { dbOpts = DBOpts {connstr = pgConn, schema = "smp_server", poolSize = 10, createSchema = True},
-          dbStoreLogPath = Nothing,
-          confirmMigrations = MCYesUp,
-          deletedTTL = 86400
-        }
-   in ServerConfig
-        { transports = [(port, transport @TLS, False)],
-          smpHandshakeTimeout = 120_000_000,
-          tbqSize = 128,
-          msgQueueQuota = 128,
-          maxJournalMsgCount = 256,
-          maxJournalStateLines = 16,
-          queueIdBytes = 24,
-          msgIdBytes = 24,
-          serverStoreCfg = SSCDatabase storeCfg,
-          storeNtfsFile = Nothing,
-          allowNewQueues = True,
-          newQueueBasicAuth = Nothing,
-          controlPortUserAuth = Nothing,
-          controlPortAdminAuth = Nothing,
-          dailyBlockQueueQuota = 20,
-          messageExpiration = Just defaultMessageExpiration,
-          expireMessagesOnStart = False,
-          expireMessagesOnSend = False,
-          idleQueueInterval = 14400,
-          notificationExpiration = defaultNtfExpiration,
-          inactiveClientExpiration = Nothing,
-          logStatsInterval = Nothing,
-          logStatsStartTime = 0,
-          serverStatsLogFile = "bench/tmp/stats.log",
-          serverStatsBackupFile = Nothing,
-          prometheusInterval = Nothing,
-          prometheusMetricsFile = "bench/tmp/metrics.txt",
-          pendingENDInterval = 500_000,
-          ntfDeliveryInterval = 200_000,
-          smpCredentials =
-            ServerCredentials
-              { caCertificateFile = Just "tests/fixtures/ca.crt",
-                privateKeyFile = "tests/fixtures/server.key",
-                certificateFile = "tests/fixtures/server.crt"
-              },
-          httpCredentials = Nothing,
-          smpServerVRange = supportedServerSMPRelayVRange,
-          Env.transportConfig = mkTransportServerConfig True (Just alpnSupportedSMPHandshakes) True,
-          controlPort = Nothing,
-          smpAgentCfg = defaultSMPClientAgentConfig {persistErrorInterval = 1},
-          allowSMPProxy = False,
-          serverClientConcurrency = 16,
-          information = Nothing,
-          startOptions = StartOptions {maintenance = False, compactLog = False, logLevel = LogInfo, skipWarnings = True, confirmMigrations = MCYesUp}
-        }
-

bench/Report.hs

@@ -1,113 +0,0 @@
-{-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE NumericUnderscores #-}
-{-# LANGUAGE OverloadedStrings #-}
-
-module Report
-  ( Snapshot (..),
-    takeSnapshot,
-    printSummary,
-    writeTimeSeriesHeader,
-    appendTimeSeries,
-  )
-where
-
-import Control.Concurrent (threadDelay)
-import Data.List (foldl')
-import Data.Text (Text)
-import qualified Data.Text as T
-import qualified Data.Text.IO as T
-import Data.Time.Clock (UTCTime, getCurrentTime)
-import Data.Time.Format.ISO8601 (iso8601Show)
-import Data.Word (Word32, Word64)
-import GHC.Stats (RTSStats (..), GCDetails (..), getRTSStats)
-import System.IO (Handle, IOMode (..), hFlush, hSetBuffering, BufferMode (..), withFile)
-import System.Mem (performMajorGC)
-
-data Snapshot = Snapshot
-  { snapTime :: UTCTime,
-    snapPhase :: Text,
-    snapLive :: Word64,
-    snapHeap :: Word64,
-    snapLarge :: Word64,
-    snapFrag :: Word64,
-    snapGCs :: Word32,
-    snapClients :: Int
-  }
-
-takeSnapshot :: Text -> Int -> IO Snapshot
-takeSnapshot phase clients = do
-  performMajorGC
-  threadDelay 1_000_000
-  rts <- getRTSStats
-  ts <- getCurrentTime
-  let GCDetails {gcdetails_live_bytes, gcdetails_mem_in_use_bytes, gcdetails_large_objects_bytes, gcdetails_block_fragmentation_bytes} = gc rts
-  pure
-    Snapshot
-      { snapTime = ts,
-        snapPhase = phase,
-        snapLive = gcdetails_live_bytes,
-        snapHeap = gcdetails_mem_in_use_bytes,
-        snapLarge = gcdetails_large_objects_bytes,
-        snapFrag = gcdetails_block_fragmentation_bytes,
-        snapGCs = gcs rts,
-        snapClients = clients
-      }
-
-printSummary :: [Snapshot] -> IO ()
-printSummary [] = putStrLn "No snapshots collected."
-printSummary snaps = do
-  putStrLn ""
-  putStrLn hdr
-  putStrLn $ replicate (length hdr) '-'
-  mapM_ printRow (zip (Snapshot {snapLive = 0, snapHeap = 0, snapLarge = 0, snapFrag = 0, snapGCs = 0, snapClients = 0, snapPhase = "", snapTime = snapTime (head snaps)} : snaps) snaps)
-  where
-    hdr = padR 20 "Phase" <> padL 12 "live_MB" <> padL 12 "large_MB" <> padL 12 "frag_MB" <> padL 12 "heap_MB" <> padL 10 "clients" <> padL 14 "d_live_MB" <> padL 14 "d_large_MB" <> padL 14 "KB/client"
-    printRow (prev, cur) =
-      putStrLn $
-        padR 20 (T.unpack $ snapPhase cur)
-          <> padL 12 (showMB $ snapLive cur)
-          <> padL 12 (showMB $ snapLarge cur)
-          <> padL 12 (showMB $ snapFrag cur)
-          <> padL 12 (showMB $ snapHeap cur)
-          <> padL 10 (show $ snapClients cur)
-          <> padL 14 (showDeltaMB (snapLive cur) (snapLive prev))
-          <> padL 14 (showDeltaMB (snapLarge cur) (snapLarge prev))
-          <> padL 14 (perClient cur)
-    showMB w = show (w `div` (1024 * 1024))
-    showDeltaMB a b
-      | a >= b = "+" <> show ((a - b) `div` (1024 * 1024))
-      | otherwise = "-" <> show ((b - a) `div` (1024 * 1024))
-    perClient Snapshot {snapClients, snapLive}
-      | snapClients > 0 = show (snapLive `div` fromIntegral snapClients `div` 1024)
-      | otherwise = "-"
-    padR n s = s <> replicate (max 0 (n - length s)) ' '
-    padL n s = replicate (max 0 (n - length s)) ' ' <> s
-
-csvHeader :: Text
-csvHeader = "timestamp,phase,rts_live,rts_heap,rts_large,rts_frag,rts_gc,clients"
-
-snapshotCsv :: Snapshot -> Text
-snapshotCsv Snapshot {snapTime, snapPhase, snapLive, snapHeap, snapLarge, snapFrag, snapGCs, snapClients} =
-  T.intercalate
-    ","
-    [ T.pack $ iso8601Show snapTime,
-      snapPhase,
-      tshow snapLive,
-      tshow snapHeap,
-      tshow snapLarge,
-      tshow snapFrag,
-      tshow snapGCs,
-      tshow snapClients
-    ]
-
-writeTimeSeriesHeader :: FilePath -> IO ()
-writeTimeSeriesHeader path = T.writeFile path (csvHeader <> "\n")
-
-appendTimeSeries :: FilePath -> Snapshot -> IO ()
-appendTimeSeries path snap =
-  withFile path AppendMode $ \h -> do
-    hSetBuffering h LineBuffering
-    T.hPutStrLn h $ snapshotCsv snap
-
-tshow :: Show a => a -> Text
-tshow = T.pack . show

bench/docker-compose.yml

@@ -1,46 +0,0 @@
-services:
-  postgres:
-    image: postgres:17
-    environment:
-      POSTGRES_USER: smp
-      POSTGRES_DB: smp_bench
-      POSTGRES_HOST_AUTH_METHOD: trust
-    volumes:
-      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
-      - pgdata:/var/lib/postgresql/data
-    healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U smp -d smp_bench"]
-      interval: 2s
-      timeout: 5s
-      retries: 10
-
-  bench:
-    build:
-      context: ..
-      dockerfile: bench/Dockerfile
-    depends_on:
-      postgres:
-        condition: service_healthy
-    environment:
-      BENCH_PG: "postgresql://smp@postgres/smp_bench"
-      BENCH_CLIENTS: "${BENCH_CLIENTS:-5000}"
-      BENCH_MINUTES: "${BENCH_MINUTES:-5}"
-    command:
-      - "--pg"
-      - "postgresql://smp@postgres/smp_bench"
-      - "--clients"
-      - "${BENCH_CLIENTS:-5000}"
-      - "--minutes"
-      - "${BENCH_MINUTES:-5}"
-      - "--timeseries"
-      - "/results/timeseries.csv"
-      - "+RTS"
-      - "-N"
-      - "-A16m"
-      - "-T"
-      - "-RTS"
-    volumes:
-      - ./results:/results
-
-volumes:
-  pgdata:

bench/init.sql

@@ -1,2 +0,0 @@
-CREATE EXTENSION IF NOT EXISTS pgcrypto;
-CREATE SCHEMA IF NOT EXISTS smp_server;

bench/run.sh

@@ -1,60 +0,0 @@
-#!/bin/bash
-set -e
-cd "$(dirname "$0")"
-
-mkdir -p results
-
-reset_db() {
-  docker compose down -v 2>/dev/null || true
-  docker compose up -d --wait postgres
-  echo "PostgreSQL ready."
-}
-
-if [ "$1" = "--compare-rts" ]; then
-  shift
-  docker compose build bench
-  for label_flags in \
-    "default:-N -A16m -T" \
-    "F1.2:-N -A16m -F1.2 -T" \
-    "F1.5:-N -A16m -F1.5 -T" \
-    "A4m:-N -A4m -T" \
-    "A4m-F1.2:-N -A4m -F1.2 -T" \
-    "compact:-N -A16m -c -T" \
-    "nonmoving:-N -A16m -xn -T"; do
-    label="${label_flags%%:*}"
-    flags="${label_flags#*:}"
-    echo ""
-    echo "=========================================="
-    echo "  RTS config: $label ($flags)"
-    echo "=========================================="
-    reset_db
-    docker compose run --rm \
-      -e BENCH_CLIENTS="${BENCH_CLIENTS:-1000}" \
-      -e BENCH_MINUTES="${BENCH_MINUTES:-2}" \
-      bench \
-      --pg "postgresql://smp@postgres/smp_bench" \
-      --clients "${BENCH_CLIENTS:-1000}" \
-      --minutes "${BENCH_MINUTES:-2}" \
-      --timeseries "/results/bench-${label}.csv" \
-      "$@" \
-      +RTS $flags -RTS
-  done
-  echo ""
-  echo "Done. Results in bench/results/"
-elif [ "$1" = "--local" ]; then
-  # Run natively (not in container) — requires local Postgres
-  shift
-  reset_db
-  cabal run smp-server-bench -f server_postgres -- \
-    --pg "postgresql://smp@localhost:15432/smp_bench" \
-    --clients "${BENCH_CLIENTS:-5000}" \
-    --minutes "${BENCH_MINUTES:-5}" \
-    "$@" \
-    +RTS -N -A16m -s -RTS
-else
-  # Run fully in containers
-  reset_db
-  docker compose run --rm bench "$@"
-fi
-
-docker compose down

docs/plans/memory-analysis-results.md

@@ -1,100 +0,0 @@
-## Memory Diagnostics Results
-
-### Data Collection
-
-Server: smp19.simplex.im, PostgreSQL backend, `useCache = False`
-RTS flags: `+RTS -N -A16m -I0.01 -Iw15 -s -RTS` (16 cores)
-
-### Mar 20 Data (1 hour, 07:19-08:19)
-
-```
-Time    rts_live  rts_heap  rts_large  rts_frag   clients  non-large
-07:19    7.5 GB    8.2 GB    5.5 GB    0.03 GB    14,000    2.0 GB
-07:24    6.4 GB   10.8 GB    5.2 GB    3.6 GB     14,806    1.2 GB
-07:29    8.2 GB   10.8 GB    6.5 GB    1.8 GB     15,667    1.7 GB
-07:34   10.0 GB   12.3 GB    7.9 GB    1.4 GB     15,845    2.1 GB
-07:39    6.7 GB   13.0 GB    5.3 GB    5.6 GB     16,589    1.4 GB
-07:44    8.5 GB   13.0 GB    6.7 GB    3.7 GB     16,283    1.8 GB
-07:49    6.5 GB   13.0 GB    5.2 GB    5.8 GB     16,532    1.3 GB
-07:54    6.0 GB   13.0 GB    4.8 GB    6.3 GB     16,636    1.2 GB
-07:59    6.4 GB   13.0 GB    5.1 GB    5.9 GB     16,769    1.3 GB
-08:04    8.3 GB   13.0 GB    6.5 GB    3.9 GB     17,352    1.8 GB
-08:09   10.2 GB   13.0 GB    8.0 GB    1.9 GB     17,053    2.2 GB
-08:14    5.6 GB   13.0 GB    4.5 GB    6.8 GB     17,147    1.1 GB
-08:19    7.6 GB   13.0 GB    6.1 GB    4.6 GB     17,496    1.5 GB
-```
-
-non-large = rts_live - rts_large (normal Haskell heap objects: Maps, TVars, closures)
-
-### Mar 19 Data (5.5 hours, 07:49-13:19)
-
-rts_heap grew from 10.1 GB to 20.7 GB over 5.5 hours.
-Post-GC rts_live floor rose from 5.5 GB to 9.1 GB.
-
-### Findings
-
-**1. Large/pinned objects dominate live data (60-80%)**
-
-`rts_large` = 4.5-8.0 GB out of 5.6-10.2 GB live. These are allocations > ~3KB that go on GHC's large object heap. They oscillate (not growing monotonically), meaning they are being allocated and freed constantly — transient, not leaked.
-
-**2. Fragmentation is the heap growth mechanism**
-
-`rts_heap ≈ rts_live + rts_frag`. The heap grows because pinned/large objects fragment GHC's block allocator. Once GHC expands the heap, it never shrinks. Growth pattern:
-- Large objects allocated → occupy blocks
-- Large objects freed → blocks can't be reused if ANY other object shares the block
-- New allocations need fresh blocks → heap expands
-- Heap never returns memory to OS
-
-**3. Non-large heap data is stable (~1.0-2.2 GB)**
-
-Normal Haskell objects (Maps, TVars, closures, client structures) account for only 1-2 GB. This scales with client count at ~100-130 KB/client and does NOT grow over time.
-
-**4. All tracked data structures are NOT the cause**
-
-- `clientSndQ=0, clientMsgQ=0` — TBQueues empty, no message accumulation
-- `smpQSubs` oscillates ~1.0-1.4M — entries are cleaned up, not leaking
-- `ntfStore` < 2K entries — negligible
-- All proxy agent maps near 0
-- `loadedQ=0` — useCache=False confirmed working
-
-**5. Source of large objects is unclear without heap profiling**
-
-The 4.5-8.0 GB of large objects could come from:
-- PostgreSQL driver (`postgresql-simple`/`libpq`) — pinned ByteStrings for query results
-- TLS library (`tls`) — pinned buffers per connection
-- Network socket I/O — pinned ByteStrings for recv/send
-- SMP protocol message blocks
-
-Cannot distinguish between these without `-hT` heap profiling (which is too expensive for this server).
-
-### Root Cause
-
-**GHC heap fragmentation from constant churn of large/pinned ByteString allocations.**
-
-Not a data structure leak. The live data itself is reasonable (5-10 GB for 15-17K clients). The problem is that GHC's copying GC cannot compact around pinned objects, so the heap grows with fragmentation and never shrinks.
-
-### Mitigation Options
-
-All are RTS flag changes — no rebuild needed, reversible by restart.
-
-**1. `-F1.2`** (reduce GC trigger factor from default 2.0)
-- Triggers major GC when heap reaches 1.2x live data instead of 2x
-- Reclaims fragmented blocks sooner
-- Trade-off: more frequent GC, slightly higher CPU
-- Risk: low — just makes GC run more often
-
-**2. Reduce `-A16m` to `-A4m`** (smaller nursery)
-- More frequent minor GC → short-lived pinned objects freed faster
-- Trade-off: more GC cycles, but each is smaller
-- Risk: low — may actually improve latency by reducing GC pause times
-
-**3. `+RTS -xn`** (nonmoving GC)
-- Designed for pinned-heavy workloads — avoids copying entirely
-- Available since GHC 8.10, improved in 9.x
-- Trade-off: different GC characteristics, less battle-tested
-- Risk: medium — different GC algorithm, should test first
-
-**4. Limit concurrent connections** (application-level)
-- Since large objects scale per-client, fewer clients = less fragmentation
-- Trade-off: reduced capacity
-- Risk: low but impacts users

docs/plans/memory-analysis.md

@@ -1,225 +0,0 @@
-## Root Cause Analysis: SMP Server Memory Growth (23.5GB)
-
-### Environment
-
-- **Server**: smp19.simplex.im, ~21,927 connected clients
-- **Storage**: PostgreSQL backend with `useCache = False`
-- **RTS flags**: `+RTS -N -A16m -I0.01 -Iw15 -s -RTS` (16 cores)
-- **Memory**: 23.5GB RES / 1031GB VIRT (75% of available RAM)
-
-### Log Summary
-
-- **Duration**: ~22 hours (Mar 16 12:12 → Mar 17 10:20)
-- **92,277 proxy connection errors** out of 92,656 total log lines (99.6%)
-- **292 unique failing destination servers**, top offender: `nowhere.moe` (12,875 errors)
-- Only **145 successful proxy connections**
-
----
-
-### Known Factor: GHC Heap Sizing
-
-With 16 cores and `-A16m`:
-- **Nursery**: 16 × 16MB = **256MB baseline**
-- GHC default major GC threshold = **2× live data** — if live data is 10GB, heap grows to ~20GB before major GC
-- The server is rarely idle with 22K clients, so major GC is deferred despite `-I0.01`
-- This is an amplifier — whatever the actual live data size is, GHC roughly doubles it
-
----
-
-### Candidate Structures That Could Grow Unboundedly
-
-Analysis of the full codebase identified these structures that either grow without bound or have uncertain cleanup:
-
-#### 1. `SubscribedClients` maps — `Env/STM.hs:378`
-
-Both `subscribers.queueSubscribers` and `ntfSubscribers.queueSubscribers` (and their `serviceSubscribers`) use `SubscribedClients (TMap EntityId (TVar (Maybe (Client s))))`.
-
-Comment at line 376: *"The subscriptions that were made at any point are not removed"*
-
-`deleteSubcribedClient` IS called on disconnect (Server.hs:1112) and DOES call `TM.delete`. But it only deletes if the current stored client matches — if another client already re-subscribed, the old client's disconnect won't remove the entry. This is by design for mobile client continuity, but the net effect on map size over time is unclear without measurement.
-
-#### 2. ProxyAgent's subscription TMaps — `Client/Agent.hs:145-151`
-
-The `SMPClientAgent` has 4 TMaps that accumulate one top-level entry per unique destination server and **never remove** them:
-
-- `activeServiceSubs :: TMap SMPServer (TVar ...)` (line 145)
-- `activeQueueSubs :: TMap SMPServer (TMap QueueId ...)` (line 146)
-- `pendingServiceSubs :: TMap SMPServer (TVar ...)` (line 149)
-- `pendingQueueSubs :: TMap SMPServer (TMap QueueId ...)` (line 150)
-
-Comment at line 262: *"these vars are never removed, they are only added"*
-
-These are only used for the proxy agent (SParty 'Sender), so they grow with each unique destination SMP server proxied to. With 292 unique servers in this log period, these are likely small — but long-running servers may accumulate thousands.
-
-`closeSMPClientAgent` (line 369) does NOT clear these 4 maps.
-
-#### 3. `NtfStore` — `NtfStore.hs:26`
-
-`NtfStore (TMap NotifierId (TVar [MsgNtf]))` — one entry per NotifierId.
-
-`deleteExpiredNtfs` (line 47) filters expired notifications from lists but does **not remove entries with empty lists** from the TMap. Over time, NotifierIds that no longer receive notifications leave zombie `TVar []` entries.
-
-`deleteNtfs` (line 44) does remove the full entry via `TM.lookupDelete` — but only called when a notifier is explicitly deleted.
-
-#### 4. `serviceLocks` in PostgresQueueStore — `Postgres.hs:112,469`
-
-`serviceLocks :: TMap CertFingerprint Lock` — one Lock per unique certificate fingerprint.
-
-`getCreateService` (line 469) calls `withLockMap (serviceLocks st) fp` which calls `getMapLock` (Agent/Client.hs:1029-1032) — this **unconditionally inserts** a Lock into the TMap. There is **no cleanup code** for serviceLocks anywhere. This is NOT guarded by `useCache`.
-
-#### 5. `sentCommands` per proxy client connection — `Client.hs:580`
-
-Each `PClient` has `sentCommands :: TMap CorrId (Request err msg)`. Entries are added per command sent (line 1369) and only removed when a response arrives (line 698). If a connection drops before all responses arrive, entries remain until the `PClient` is GC'd. Since `PClient` is captured by the connection thread which terminates on error, the `PClient` should become GC-eligible — but GC timing depends on heap pressure.
-
-#### 6. `subQ :: TQueue (ClientSub, ClientId)` — `Env/STM.hs:363`
-
-Unbounded `TQueue` for subscription changes. If the subscriber thread (`serverThread`) can't process changes fast enough, this queue grows without backpressure. With 22K clients subscribing/unsubscribing, sustained bursts could cause this queue to bloat.
-
----
-
-### Ruled Out
-
-1. **PostgreSQL queue cache**: `useCache = False` — `queues`, `senders`, `links`, `notifiers` TMaps are empty.
-2. **`notifierLocks`**: Guarded by `useCache` (Postgres.hs:377,405) — not used with `useCache = False`.
-3. **Client structures**: 22K × ~3KB = ~66MB — negligible.
-4. **TBQueues**: Bounded (`tbqSize = 128`).
-5. **Thread management**: `forkClient` uses weak refs + `finally` blocks. `endThreads` cleared on disconnect.
-6. **Proxy `smpClients`/`smpSessions`**: Properly cleaned on disconnect/expiry.
-7. **`smpSubWorkers`**: Properly cleaned on worker completion; also cleared in `closeSMPClientAgent`.
-8. **`pendingEvents`**: Atomically swapped empty every `pendingENDInterval`.
-9. **Stats IORef counters**: Fixed number, bounded.
-10. **DB connection pool**: Bounded `TBQueue` with bracket-based return.
-
----
-
-### Insufficient Data to Determine Root Cause
-
-Without measuring the actual sizes of these structures at runtime, we cannot determine which (if any) is the primary contributor. The following exact logging changes will identify the root cause.
-
----
-
-### EXACT LOGS TO ADD
-
-Add a new periodic logging thread in `src/Simplex/Messaging/Server.hs`.
-
-Insert at `Server.hs:197` (after `prometheusMetricsThread_`):
-
-```haskell
-        <> memoryDiagThread_ cfg
-```
-
-Then define:
-
-```haskell
-    memoryDiagThread_ :: ServerConfig s -> [M s ()]
-    memoryDiagThread_ ServerConfig {prometheusInterval = Just _} =
-      [memoryDiagThread]
-    memoryDiagThread_ _ = []
-
-    memoryDiagThread :: M s ()
-    memoryDiagThread = do
-      labelMyThread "memoryDiag"
-      Env { ntfStore = NtfStore ntfMap
-          , server = srv@Server {subscribers, ntfSubscribers}
-          , proxyAgent = ProxyAgent {smpAgent = pa}
-          , msgStore_ = ms
-          } <- ask
-      let interval = 300_000_000 -- 5 minutes
-      liftIO $ forever $ do
-        threadDelay interval
-        -- GHC RTS stats
-        rts <- getRTSStats
-        let liveBytes = gcdetails_live_bytes $ gc rts
-            heapSize = gcdetails_mem_in_use_bytes $ gc rts
-            gcCount = gcs rts
-        -- Server structures
-        clientCount <- IM.size <$> getServerClients srv
-        -- SubscribedClients (queue and service subscribers for both SMP and NTF)
-        smpQSubs <- M.size <$> getSubscribedClients (queueSubscribers subscribers)
-        smpSSubs <- M.size <$> getSubscribedClients (serviceSubscribers subscribers)
-        ntfQSubs <- M.size <$> getSubscribedClients (queueSubscribers ntfSubscribers)
-        ntfSSubs <- M.size <$> getSubscribedClients (serviceSubscribers ntfSubscribers)
-        -- Pending events
-        smpPending <- IM.size <$> readTVarIO (pendingEvents subscribers)
-        ntfPending <- IM.size <$> readTVarIO (pendingEvents ntfSubscribers)
-        -- NtfStore
-        ntfStoreSize <- M.size <$> readTVarIO ntfMap
-        -- ProxyAgent maps
-        let SMPClientAgent {smpClients, smpSessions, activeServiceSubs, activeQueueSubs, pendingServiceSubs, pendingQueueSubs, smpSubWorkers} = pa
-        paClients <- M.size <$> readTVarIO smpClients
-        paSessions <- M.size <$> readTVarIO smpSessions
-        paActSvc <- M.size <$> readTVarIO activeServiceSubs
-        paActQ <- M.size <$> readTVarIO activeQueueSubs
-        paPndSvc <- M.size <$> readTVarIO pendingServiceSubs
-        paPndQ <- M.size <$> readTVarIO pendingQueueSubs
-        paWorkers <- M.size <$> readTVarIO smpSubWorkers
-        -- Loaded queue counts
-        lc <- loadedQueueCounts $ fromMsgStore ms
-        -- Log everything
-        logInfo $
-          "MEMORY "
-            <> "rts_live=" <> tshow liveBytes
-            <> " rts_heap=" <> tshow heapSize
-            <> " rts_gc=" <> tshow gcCount
-            <> " clients=" <> tshow clientCount
-            <> " smpQSubs=" <> tshow smpQSubs
-            <> " smpSSubs=" <> tshow smpSSubs
-            <> " ntfQSubs=" <> tshow ntfQSubs
-            <> " ntfSSubs=" <> tshow ntfSSubs
-            <> " smpPending=" <> tshow smpPending
-            <> " ntfPending=" <> tshow ntfPending
-            <> " ntfStore=" <> tshow ntfStoreSize
-            <> " paClients=" <> tshow paClients
-            <> " paSessions=" <> tshow paSessions
-            <> " paActSvc=" <> tshow paActSvc
-            <> " paActQ=" <> tshow paActQ
-            <> " paPndSvc=" <> tshow paPndSvc
-            <> " paPndQ=" <> tshow paPndQ
-            <> " paWorkers=" <> tshow paWorkers
-            <> " loadedQ=" <> tshow (loadedQueueCount lc)
-            <> " loadedNtf=" <> tshow (loadedNotifierCount lc)
-            <> " ntfLocks=" <> tshow (notifierLockCount lc)
-```
-
-Note: `smpSubs.subsCount` (queueSubscribers size) and `smpSubs.subServicesCount` (serviceSubscribers size) are **already logged** in Prometheus (lines 475-496). The log above adds all other candidate structures plus GHC RTS memory stats.
-
-This produces a single log line every 5 minutes:
-
-```
-[INFO] MEMORY rts_live=10737418240 rts_heap=23488102400 rts_gc=4521 clients=21927 smpQSubs=1847233 smpSSubs=42 ntfQSubs=982112 ntfSSubs=31 smpPending=0 ntfPending=0 ntfStore=512844 paClients=12 paSessions=12 paActSvc=0 paActQ=0 paPndSvc=0 paPndQ=0 paWorkers=3 loadedQ=0 loadedNtf=0 ntfLocks=0
-```
-
-### What Each Metric Tells Us
-
-| Metric | What it reveals | If growing = suspect |
-|--------|----------------|---------------------|
-| `rts_live` | Actual live data after last major GC | Baseline — everything else should add up to this |
-| `rts_heap` | Total heap (should be ~2× rts_live) | If >> 2× live, fragmentation issue |
-| `clients` | Connected client count | Known: ~22K |
-| `smpQSubs` | SubscribedClients map size (queue subs) | If >> clients × avg_subs, entries not cleaned |
-| `smpSSubs` | SubscribedClients map size (service subs) | Should be small |
-| `ntfQSubs` | NTF SubscribedClients map (queue subs) | Same concern as smpQSubs |
-| `ntfSSubs` | NTF SubscribedClients map (service subs) | Should be small |
-| `smpPending` / `ntfPending` | Pending END/DELD events per client | If large, subscriber thread lagging |
-| `ntfStore` | NotifierId count in NtfStore | If growing monotonically, zombie entries |
-| `paClients` | Proxy connections to other servers | Should be <= unique dest servers |
-| `paSessions` | Active proxy sessions | Should match paClients |
-| `paActSvc` / `paActQ` | Proxy active subscriptions | If growing, entries never removed |
-| `paPndSvc` / `paPndQ` | Proxy pending subscriptions | If growing, resubscription stuck |
-| `paWorkers` | Active reconnect workers | If growing, workers stuck in retry |
-| `loadedQ` | Cached queues in store (0 with useCache=False) | Should be 0 |
-| `ntfLocks` | Notifier locks in store | Should be 0 with useCache=False |
-
-### Interpretation Guide
-
-**If `smpQSubs` is in the millions**: SubscribedClients is the primary leak. Entries accumulate for every queue ever subscribed to.
-
-**If `ntfStore` grows monotonically**: Zombie notification entries (empty lists after expiration). Fix: `deleteExpiredNtfs` should remove entries with empty lists.
-
-**If `paActSvc` + `paActQ` grow**: Proxy agent subscription maps are the leak. Fix: add cleanup when no active/pending subs exist for a server.
-
-**If `rts_live` is much smaller than `rts_heap`**: GHC heap fragmentation. Fix: tune `-F` flag (GC trigger factor) or use `-c` (compacting GC).
-
-**If `rts_live` ~ 10-12GB**: The live data is genuinely large. Look at which metric is the largest contributor.
-
-**If nothing above is large but `rts_live` is large**: The leak is in a structure not measured here — likely TLS connection buffers, ByteString retention from Postgres queries, or GHC runtime overhead. Next step would be heap profiling with `-hT`.

plans/slow-queries-analysis-2.md

@@ -0,0 +1,184 @@
+# SMP Server Postgres: Slow Query Analysis
+
+Data from three production servers (A, B, C), ~3.5 day observation window.
+
+## Top queries by total time
+
+| Rank | Query | Server A ms | Server B ms | Server C ms |
+|------|-------|-------------|-------------|-------------|
+| 1 | getEntityCounts (6 COUNT subqueries) | 1,682,874 | 1,639,325 | 1,619,892 |
+| 2 | write_message() | 303,393 | 458,262 | 280,375 |
+| 3 | try_del_peek_msg() | 352,912 | 386,036 | 333,877 |
+| 4 | expire_old_messages() | 246,034 | 220,003 | 160,232 |
+| 5 | UPDATE SET updated_at | 234,146 | 216,911 | 211,480 |
+| 6 | INSERT INTO messages | 184,430 | 323,617 | 169,149 |
+| 7 | expire batch cursor (array_agg) | 122,739 | 99,061 | 39,975 |
+| 8 | Batch recipient_id IN lookups | ~134K | ~79K | ~81K |
+| 9 | Batch notifier_id IN lookups | ~143K | ~64K | ~64K |
+| 10 | msg_peek (SELECT FROM messages) | ~112K | ~102K | ~126K |
+
+getEntityCounts alone is **45-48%** of total query time on all three servers.
+
+---
+
+## Verified fixes
+
+### 1. getEntityCounts: replace ③④ with SUM(queue_count)
+
+**Query** (`QueueStore/Postgres.hs:160-167`):
+
+```sql
+SELECT
+  (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL) AS queue_count,                                            -- ①
+  (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL AND notifier_id IS NOT NULL) AS notifier_count,              -- ②
+  (SELECT COUNT(1) FROM services WHERE service_role = ?) AS rcv_service_count,                                           -- trivial
+  (SELECT COUNT(1) FROM services WHERE service_role = ?) AS ntf_service_count,                                           -- trivial
+  (SELECT COUNT(1) FROM msg_queues WHERE rcv_service_id IS NOT NULL AND deleted_at IS NULL) AS rcv_service_queues_count,  -- ③
+  (SELECT COUNT(1) FROM msg_queues WHERE ntf_service_id IS NOT NULL AND deleted_at IS NULL) AS ntf_service_queues_count   -- ④
+```
+
+| Server | Calls | Avg ms | Max ms | Total ms |
+|--------|-------|--------|--------|----------|
+| A | 5,058 | 332.7 | 2,061 | 1,682,874 |
+| B | 5,055 | 324.3 | 1,844 | 1,639,325 |
+| C | 5,053 | 320.6 | 1,250 | 1,619,892 |
+
+**Problem**: 4 subqueries scan `msg_queues`. Indexes exist for ③
+(`idx_msg_queues_rcv_service_id(rcv_service_id, deleted_at)`), ④
+(`idx_msg_queues_ntf_service_id(ntf_service_id, deleted_at)`), and potentially ①
+(`idx_msg_queues_updated_at_recipient_id(deleted_at, ...)`), but actual query plans
+and per-subquery cost are unknown without `EXPLAIN ANALYZE`.
+
+**Fix**: Replace ③ and ④:
+
+```sql
+COALESCE((SELECT SUM(queue_count) FROM services WHERE service_role = 'M'), 0) AS rcv_service_queues_count,
+COALESCE((SELECT SUM(queue_count) FROM services WHERE service_role = 'N'), 0) AS ntf_service_queues_count
+```
+
+**Verification**:
+- Trigger logic traced for all transitions (NULL→set, change, soft-delete, physical delete) — correct.
+- FK `rcv_service_id REFERENCES services(service_id)` guarantees equivalence.
+- `queue_count + p_change` is atomic under READ COMMITTED.
+- `update_all_aggregates()` exists as repair mechanism.
+
+**Savings**: Eliminates 2 of 4 msg_queues scans. Exact per-subquery cost unknown — needs `EXPLAIN ANALYZE`.
+
+---
+
+### 2. expire_old_messages: remove trailing COUNTs
+
+At the end of `expire_old_messages` (`Migrations.hs`):
+
+```sql
+r_stored_msgs_count := (SELECT COUNT(1) FROM messages);
+r_stored_queues := (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL);
+```
+
+| Server | expire avg | r_stored_queues avg | r_stored_msgs avg | COUNTs combined | % of procedure |
+|--------|-----------|---------------------|-------------------|-----------------|----------------|
+| A | 11,716ms | 695ms | 110ms | 805ms | 6.9% |
+| B | 10,476ms | 631ms | 17ms | 648ms | 6.2% |
+| C | 7,630ms | 588ms | 53ms | 641ms | 8.4% |
+
+**How used** (`Server.hs:485-488`):
+- `storedMsgsCount` → resets `msgCount` stat (also maintained incrementally: +1 on send, -1 on ACK)
+- `storedQueues` → **only logged** via `printMessageStats`. Same value available from `getEntityCounts`.
+
+**Fix**: Remove both COUNTs. Return only `r_expired_msgs_count`.
+
+**Verification**: All usages traced. `storedQueues` is only logged. `storedMsgsCount` resets
+an incrementally-maintained counter — removing means potential drift, corrected on restart.
+
+**Savings**: 641–805ms per cycle × 21 cycles = **13.5–16.9s total** (CSV-verified).
+
+---
+
+### 3. Trigger WHEN clause: skip PL/pgSQL call for non-service updates
+
+**Current** (`Migrations.hs:566-568`):
+
+```sql
+CREATE TRIGGER tr_queue_update
+AFTER UPDATE ON msg_queues
+FOR EACH ROW EXECUTE PROCEDURE on_queue_update();
+```
+
+Server C data — ~2.6M updates, only ~110K (4.3%) change service fields:
+
+| UPDATE pattern | Calls | Service fields? |
+|----------------|-------|-----------------|
+| SET updated_at | ~1,275K | No |
+| SET msg_can_write/size/expire (write_message) | ~600K | No |
+| SET msg_can_write/size/expire (try_del_*) | ~331K | No |
+| SET msg_can_write/size/expire (try_del reset) | ~258K | No |
+| SET sender_key | ~17K | No |
+| SET msg_can_write/size/expire (delete_expired) | ~3K | No |
+| **SET rcv_service_id** | **~101K** | **Yes** |
+| **SET deleted_at** | **~10K** | **Yes** |
+
+**Fix**: Add `WHEN` clause — evaluated in C by PostgreSQL, skips function call entirely:
+
+```sql
+CREATE TRIGGER tr_queue_update
+AFTER UPDATE ON msg_queues
+FOR EACH ROW
+WHEN (
+  OLD.deleted_at IS DISTINCT FROM NEW.deleted_at
+  OR OLD.rcv_service_id IS DISTINCT FROM NEW.rcv_service_id
+  OR OLD.ntf_service_id IS DISTINCT FROM NEW.ntf_service_id
+  OR OLD.notifier_id IS DISTINCT FROM NEW.notifier_id
+)
+EXECUTE PROCEDURE on_queue_update();
+```
+
+**Verification**:
+- PostgreSQL supports `OLD`/`NEW` in `WHEN` for `AFTER UPDATE` triggers.
+- 4 conditions match exactly the fields checked inside `on_queue_update()`.
+- Behavioral change: **none**.
+
+**Savings**: ~2.5M PL/pgSQL calls avoided. Per-call overhead estimated ~0.02–0.05ms.
+Total: ~50–125s estimated, not measured.
+
+---
+
+## Needs EXPLAIN ANALYZE
+
+### 4. Partial indexes for getEntityCounts ① and ②
+
+After fix #1, subqueries ① and ② remain. ② has no index covering both
+`deleted_at IS NULL` and `notifier_id IS NOT NULL`.
+
+```sql
+CREATE INDEX idx_msg_queues_live ON msg_queues ((1)) WHERE deleted_at IS NULL;
+CREATE INDEX idx_msg_queues_live_notifier ON msg_queues ((1)) WHERE deleted_at IS NULL AND notifier_id IS NOT NULL;
+```
+
+**Trade-off**: Write overhead on every INSERT/UPDATE/DELETE. Need `EXPLAIN ANALYZE`
+to confirm PostgreSQL uses these for COUNT vs choosing seq scan.
+
+---
+
+## Not problems
+
+- **write_message / try_del_peek_msg** (ranks #2-3): 0.5-0.7ms avg. High total from volume (~600K calls).
+  Max spikes (490-523ms) are lock contention on `FOR UPDATE` — architectural, not fixable.
+
+- **UPDATE SET updated_at** (rank #5): 0.17ms avg, ~1.3M calls. Already minimal.
+  Fix #3 eliminates the trigger overhead on these.
+
+- **SET rcv_service_id** (Server C only, 100K calls): CSV shows rows_affected = calls — all
+  legitimate associations. Haskell guard at `Postgres.hs:487` works correctly.
+
+- **Batch lookups** (ranks #8-9): 1.8-2.0ms avg for ~135 PK probes. Near-optimal.
+
+---
+
+## Summary
+
+| # | Fix | Per-call savings | Calls | Total savings | Verified |
+|---|-----|-----------------|-------|---------------|----------|
+| 1 | getEntityCounts ③④ → `SUM(queue_count)` | 0–158ms (unknown split) | ~5,050 | 0–800s | Correctness: yes. Savings: needs EXPLAIN ANALYZE |
+| 2 | Remove trailing COUNTs from expire_old_messages | 641–805ms | 21 | 13.5–16.9s | Yes (CSV) |
+| 3 | Add WHEN clause to tr_queue_update | ~0.02–0.05ms | ~2.5M skipped | ~50–125s est. | Correctness: yes. Savings: estimated |
+| 4 | Partial indexes for ①② | Unknown | ~5,050 | Unknown | Needs EXPLAIN ANALYZE |

plans/slow-queries-analysis-3.md

@@ -0,0 +1,186 @@
+# SMP Server Postgres: Slow Query Analysis (post-reset)
+
+Data from three production servers (A, B, C), ~5.5 hour window after stats reset.
+EXPLAIN ANALYZE from a large server (~30M rows in msg_queues).
+
+## Top queries by total time
+
+| Rank | Query | A total ms | B total ms | C total ms |
+|------|-------|-----------|-----------|-----------|
+| 1 | getEntityCounts (6 COUNT subqueries) | 99,799 | 101,946 | 121,495 |
+| 2 | UPDATE SET updated_at | 35,799 | 30,797 | 26,551 |
+| 3 | try_del_peek_msg() | 28,903 | 23,147 | 26,482 |
+| 4 | write_message() | 23,103 | 18,942 | 20,301 |
+| 5 | Batch recipient_id IN lookups | 16,062 | 12,529 | 9,977 |
+| 6 | INSERT INTO messages | 14,395 | 11,911 | 12,882 |
+| 7 | msg_peek (SELECT FROM messages) | 12,882 | 11,992 | 14,661 |
+| 8 | expire_old_messages() | 9,762 | 11,619 | 10,863 |
+| 9 | delete_expired_msgs() | 7,801 | 5,456 | 6,256 |
+| 10 | expire batch cursor (array_agg) | 4,421 | 5,679 | 4,566 |
+
+Grand totals: A ~292s, B ~317s, C ~288s.
+
+getEntityCounts is **34-42%** of all query time across all three servers.
+
+---
+
+## EXPLAIN ANALYZE results for getEntityCounts
+
+Run on a large server (~30M rows in msg_queues, cold cache):
+
+| Subquery | Time | % of 23.4s | Plan | Rows scanned | Key detail |
+|----------|------|-----------|------|-------------|------------|
+| ① queue_count | **7,851ms** | **33.5%** | Parallel Seq Scan | 30.6M (97% match) | No useful index |
+| ② notifier_count | **6,382ms** | **27.2%** | Parallel Seq Scan | 30.6M (4M match) | No useful index |
+| ③ rcv_service_queues | 0.5ms | 0% | Index Only Scan | 0 rows | No rcv services on this server; similar cost to ④ on servers with rcv services |
+| ④ ntf_service_queues | **8,914ms** | **38.0%** | Parallel Index Only Scan | 3.7M match | **2.7M heap fetches** |
+| Services (③④) | 1.8ms | 0% | Index Only / Bitmap | 0 + 6 rows | Trivial |
+| JIT + Planning | 1,341ms | 5.7% | — | — | JIT compilation overhead |
+| **Total** | **23,440ms** | | | | |
+
+The CSV averages (302-368ms) reflect warm-cache performance. This EXPLAIN is cold cache
+(`shared read=3.4M` vs `shared hit=44K` — 98.7% read from disk).
+
+---
+
+## Verified fixes
+
+### 1. getEntityCounts: replace ③④ with SUM(queue_count)
+
+```sql
+-- Current ③ and ④:
+(SELECT COUNT(1) FROM msg_queues WHERE rcv_service_id IS NOT NULL AND deleted_at IS NULL)  -- ③: 0.5ms
+(SELECT COUNT(1) FROM msg_queues WHERE ntf_service_id IS NOT NULL AND deleted_at IS NULL)  -- ④: 8,914ms
+
+-- Fix:
+COALESCE((SELECT SUM(queue_count) FROM services WHERE service_role = 'M'), 0)  -- ~0ms
+COALESCE((SELECT SUM(queue_count) FROM services WHERE service_role = 'N'), 0)  -- ~0ms
+```
+
+**EXPLAIN ANALYZE confirmed**: ③ returns 0 rows (already free). ④ costs 8,914ms due to
+Parallel Index Only Scan with 2.7M heap fetches on `idx_msg_queues_ntf_service_id`.
+
+**Verification**:
+- Trigger logic traced for all transitions (NULL→set, change, soft-delete, physical delete) — correct.
+- FK `rcv_service_id REFERENCES services(service_id)` guarantees equivalence.
+- `queue_count + p_change` is atomic under READ COMMITTED.
+- `update_all_aggregates()` exists as repair mechanism.
+
+**Savings**: ~8.9s cold cache (38% of query). Warm cache proportionally less but still dominant ④ cost.
+
+---
+
+### 2. getEntityCounts: partial indexes for ① and ②
+
+```sql
+CREATE INDEX idx_msg_queues_active ON msg_queues ((1)) WHERE deleted_at IS NULL;
+CREATE INDEX idx_msg_queues_active_notifier ON msg_queues ((1)) WHERE deleted_at IS NULL AND notifier_id IS NOT NULL;
+```
+
+**EXPLAIN ANALYZE confirmed**: ① does Parallel Seq Scan (7,851ms), ② does Parallel Seq Scan (6,382ms).
+No existing index is used for these subqueries despite `idx_msg_queues_updated_at_recipient_id`
+having `deleted_at` as first column — PostgreSQL chose seq scan because 97% of rows match.
+
+Partial indexes contain only matching rows, enabling fast index-only COUNT without scanning
+the full table.
+
+**Trade-off**: Write overhead on every INSERT/UPDATE/DELETE that changes `deleted_at` or
+`notifier_id`. For the ~30M row table with ~300K updates per 5.5h, this is acceptable.
+
+**Savings**: ~14.2s cold cache (61% of query). Combined with fix #1: **23.1s → ~1.3s** (JIT only).
+
+---
+
+### 3. expire_old_messages: remove trailing COUNTs
+
+At the end of `expire_old_messages` (`Migrations.hs`):
+
+```sql
+r_stored_msgs_count := (SELECT COUNT(1) FROM messages);
+r_stored_queues := (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL);
+```
+
+| Server | expire avg | r_stored_queues | r_stored_msgs | COUNTs combined | % of procedure |
+|--------|-----------|-----------------|---------------|-----------------|----------------|
+| A | 9,762ms | 535ms | 50ms | 585ms | 6.0% |
+| B | 11,619ms | 390ms | 12ms | 402ms | 3.5% |
+| C | 10,863ms | 659ms | 25ms | 684ms | 6.3% |
+
+**Usage** (`Server.hs:485-488`):
+- `storedMsgsCount` → resets `msgCount` stat (also maintained incrementally: +1 on send, -1 on ACK).
+- `storedQueues` → **only logged**. Same value available from `getEntityCounts`.
+
+**Fix**: Remove both COUNTs. Return only `r_expired_msgs_count`.
+
+**Verification**: All usages traced. `storedQueues` is only logged. `storedMsgsCount` resets
+an incrementally-maintained counter — removing means potential drift, corrected on restart.
+
+**Savings**: 402–684ms per cycle (CSV-verified).
+
+---
+
+### 4. Trigger WHEN clause: skip PL/pgSQL call for non-service updates
+
+**Current** (`Migrations.hs:566-568`):
+
+```sql
+CREATE TRIGGER tr_queue_update
+AFTER UPDATE ON msg_queues
+FOR EACH ROW EXECUTE PROCEDURE on_queue_update();
+```
+
+Server C data — ~297K updates, only ~670 (0.2%) change service-related fields:
+
+| UPDATE pattern | Calls | Service fields? |
+|----------------|-------|-----------------|
+| SET updated_at | ~213K | No |
+| SET msg_can_write/size/expire (write_message) | ~40K | No |
+| SET msg_can_write/size/expire (try_del_*) | ~27K | No |
+| SET msg_can_write/size (try_del reset) | ~14K | No |
+| SET sender_key | ~2K | No |
+| **SET deleted_at** | **~563** | **Yes** |
+| **ntf_service_id/notifier_id changes** | **~108** | **Yes** |
+
+**Fix**: Add `WHEN` clause — evaluated in C by PostgreSQL, skips function call entirely:
+
+```sql
+CREATE TRIGGER tr_queue_update
+AFTER UPDATE ON msg_queues
+FOR EACH ROW
+WHEN (
+  OLD.deleted_at IS DISTINCT FROM NEW.deleted_at
+  OR OLD.rcv_service_id IS DISTINCT FROM NEW.rcv_service_id
+  OR OLD.ntf_service_id IS DISTINCT FROM NEW.ntf_service_id
+  OR OLD.notifier_id IS DISTINCT FROM NEW.notifier_id
+)
+EXECUTE PROCEDURE on_queue_update();
+```
+
+**Verification**:
+- PostgreSQL supports `OLD`/`NEW` in `WHEN` for `AFTER UPDATE` triggers.
+- 4 conditions match exactly the fields checked inside `on_queue_update()`.
+- Behavioral change: **none**.
+
+**Savings**: ~296K PL/pgSQL calls avoided (99.8% of trigger fires). Per-call overhead
+estimated ~0.02–0.05ms. Total: ~6–15s estimated over this 5.5h window.
+
+---
+
+## Not problems
+
+- **write_message / try_del_peek_msg**: 0.5-0.65ms avg. High total from volume. Max spikes are lock contention — architectural.
+- **UPDATE SET updated_at**: 0.12-0.17ms avg. Fix #4 eliminates trigger overhead.
+- **Batch lookups**: 1.7-2.6ms avg for ~135 PK probes. Near-optimal.
+
+---
+
+## Summary
+
+| # | Fix | Savings | Verified |
+|---|-----|---------|----------|
+| 1 | getEntityCounts ③④ → `SUM(queue_count)` | ~8.9s/call cold, ④ eliminated (EXPLAIN ANALYZE) | Yes |
+| 2 | Partial indexes for getEntityCounts ①② | ~14.2s/call cold, ①② eliminated (EXPLAIN ANALYZE) | Yes — plan confirmed, index benefit to verify after creation |
+| 3 | Remove trailing COUNTs from expire_old_messages | 402–684ms/cycle (CSV) | Yes |
+| 4 | Add WHEN clause to tr_queue_update | ~6–15s est. over 5.5h (296K calls skipped) | Correctness: yes. Savings: estimated |
+
+Fixes #1 + #2 combined: getEntityCounts **23.4s → ~1.3s cold cache** (94% reduction).

plans/slow-queries-analysis.md

@@ -0,0 +1,199 @@
+# SMP Server Postgres: Slow Query Analysis
+
+Data from three production servers (A, B, C) over a multi-day observation window.
+
+## Verified fixes
+
+### 1. getEntityCounts: replace ③④ with SUM(queue_count)
+
+**Query** (`QueueStore/Postgres.hs:160-167`):
+
+```sql
+SELECT
+  (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL) AS queue_count,                                            -- ① scan
+  (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL AND notifier_id IS NOT NULL) AS notifier_count,              -- ② scan
+  (SELECT COUNT(1) FROM services WHERE service_role = ?) AS rcv_service_count,                                           -- trivial (<10 rows)
+  (SELECT COUNT(1) FROM services WHERE service_role = ?) AS ntf_service_count,                                           -- trivial (<10 rows)
+  (SELECT COUNT(1) FROM msg_queues WHERE rcv_service_id IS NOT NULL AND deleted_at IS NULL) AS rcv_service_queues_count,  -- ③ scan
+  (SELECT COUNT(1) FROM msg_queues WHERE ntf_service_id IS NOT NULL AND deleted_at IS NULL) AS ntf_service_queues_count   -- ④ scan
+```
+
+**Performance**: ~315ms avg, ~2s max, ~2500 calls. #1 slow query by total time (~800s).
+
+**Problem**: 4 scans of `msg_queues`. Indexes exist for some subqueries
+(`idx_msg_queues_rcv_service_id(rcv_service_id, deleted_at)` for ③,
+`idx_msg_queues_ntf_service_id(ntf_service_id, deleted_at)` for ④,
+`idx_msg_queues_updated_at_recipient_id(deleted_at, ...)` potentially for ①),
+but whether PostgreSQL uses them for COUNT and the actual per-subquery cost is
+unknown without `EXPLAIN ANALYZE`.
+
+**Fix**: Replace ③ and ④:
+
+```sql
+COALESCE((SELECT SUM(queue_count) FROM services WHERE service_role = 'M'), 0) AS rcv_service_queues_count,
+COALESCE((SELECT SUM(queue_count) FROM services WHERE service_role = 'N'), 0) AS ntf_service_queues_count
+```
+
+**Verification**:
+- Trigger logic traced for all transitions (NULL→set, change, soft-delete, physical delete) — correct.
+- FK `rcv_service_id REFERENCES services(service_id)` guarantees every non-NULL value maps to a row.
+- `queue_count + p_change` is atomic under READ COMMITTED — concurrent-safe.
+- `update_all_aggregates()` exists as repair mechanism.
+- `services` table has <10 rows — SUM is O(1) vs full table scan.
+
+**Savings**: Eliminates 2 of 4 msg_queues scans (③④ → trivial SUM on <10 rows).
+Exact savings unknown — if ③④ already use indexes efficiently, savings may be modest.
+`EXPLAIN ANALYZE` needed to measure actual per-subquery cost.
+
+---
+
+### 2. expire_old_messages: remove trailing COUNTs
+
+**Stored procedure** (`Migrations.hs`), at the end of `expire_old_messages`:
+
+```sql
+r_expired_msgs_count := total_deleted;
+r_stored_msgs_count := (SELECT COUNT(1) FROM messages);           -- 13-114ms per call (CSV)
+r_stored_queues := (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL);  -- 544-719ms per call (CSV)
+```
+
+**Performance**: 10 calls per observation window. Per-call cost of these COUNTs (from CSV):
+
+| Server | expire_old_messages avg | r_stored_queues avg | r_stored_msgs avg | COUNTs combined | % of procedure |
+|---|---|---|---|---|---|
+| A | 11,798ms | 719ms | 114ms | 833ms | 7.1% |
+| B | 11,242ms | 544ms | 13ms | 557ms | 5.0% |
+| C | 7,296ms | 588ms | 67ms | 655ms | 9.0% |
+
+**How results are used** (`Server.hs:485-488`):
+
+```haskell
+Right msgStats@MessageStats {storedMsgsCount = stored, expiredMsgsCount = expired} -> do
+  atomicWriteIORef (msgCount stats) stored     -- resets msgCount from storedMsgsCount
+  atomicModifyIORef'_ (msgExpired stats) (+ expired)
+  printMessageStats "STORE: messages" msgStats  -- logs all three fields
+```
+
+- `expiredMsgsCount` — computed incrementally in the loop. **Needed, already cheap.**
+- `storedMsgsCount` — used to reset `msgCount` stat. But `msgCount` is also maintained
+  incrementally (`+1` on send at line 1963, `-1` on ACK at line 1916). The reset corrects drift.
+- `storedQueues` — **used only for logging**. Same value available from `getEntityCounts`
+  which runs every ~60s via Prometheus.
+
+**Fix**: Remove both COUNTs from the stored procedure. Return only `r_expired_msgs_count`.
+
+For `storedMsgsCount`: either trust the incremental `msgCount` counter, or query
+`SELECT COUNT(1) FROM messages` separately (in parallel, not blocking the procedure).
+
+For `storedQueues`: use the value from the most recent `getEntityCounts` call.
+
+**Verification**: Traced all usages of `MessageStats` fields from `expireOldMessages` in
+`Server.hs`. `storedQueues` is only logged. `storedMsgsCount` resets a counter that's already
+maintained incrementally — removing the reset means potential drift, but the counter is
+corrected on next server restart anyway.
+
+**Savings**: 560-830ms per expiration cycle × 10 cycles = **5.6-8.3s total** over observation window.
+
+---
+
+### 3. Trigger WHEN clause: skip function call for non-service updates
+
+**Current trigger** (`Migrations.hs:566-568`):
+
+```sql
+CREATE TRIGGER tr_queue_update
+AFTER UPDATE ON msg_queues
+FOR EACH ROW EXECUTE PROCEDURE on_queue_update();
+```
+
+Fires on **every UPDATE** to `msg_queues`. From Server C data, ~1.2M updates per observation
+window, but only ~105K (8.7%) actually change service-related fields:
+
+| UPDATE pattern | Calls | Changes service fields? |
+|---|---|---|
+| SET updated_at | ~427K | No |
+| SET msg_can_write/size/expire (write_message) | ~336K | No |
+| SET msg_can_write/size/expire (try_del_*) | ~196K | No |
+| SET msg_can_write/size/expire (delete_expired) | ~136K | No |
+| SET sender_key | ~8K | No |
+| SET status | ~2K | No |
+| **SET rcv_service_id** | **~101K** | **Yes** |
+| **SET deleted_at** | **~5K** | **Yes** |
+
+The `on_queue_update()` PL/pgSQL function evaluates 8-12 boolean conditions on every call,
+then returns without calling `update_aggregates` for 91% of invocations.
+
+**Fix**: Add a `WHEN` clause to the trigger definition. PostgreSQL evaluates `WHEN` in C code
+before calling the PL/pgSQL function — no function entry overhead at all:
+
+```sql
+CREATE TRIGGER tr_queue_update
+AFTER UPDATE ON msg_queues
+FOR EACH ROW
+WHEN (
+  OLD.deleted_at IS DISTINCT FROM NEW.deleted_at
+  OR OLD.rcv_service_id IS DISTINCT FROM NEW.rcv_service_id
+  OR OLD.ntf_service_id IS DISTINCT FROM NEW.ntf_service_id
+  OR OLD.notifier_id IS DISTINCT FROM NEW.notifier_id
+)
+EXECUTE PROCEDURE on_queue_update();
+```
+
+**Verification**:
+- PostgreSQL supports `OLD`/`NEW` in `WHEN` clauses for `AFTER UPDATE` triggers.
+- The 4 conditions match exactly the fields checked inside `on_queue_update()`.
+- When WHEN is false, the function is **never called** — zero PL/pgSQL overhead.
+- When WHEN is true, the function runs identically to today.
+- Behavioral change: **none** — same aggregates updated in same cases.
+
+**Savings**: ~1.1M PL/pgSQL function calls avoided. Each call has fixed overhead
+(function entry, OLD/NEW row extraction, condition evaluation, return). Exact savings
+need measurement, but function call overhead is non-trivial at this volume.
+
+---
+
+## Fixes that need EXPLAIN ANALYZE
+
+### 4. Partial indexes for getEntityCounts ① and ②
+
+Subqueries ① and ② still scan msg_queues. ① may use
+`idx_msg_queues_updated_at_recipient_id(deleted_at, ...)` but ② has no index covering
+both `deleted_at IS NULL` and `notifier_id IS NOT NULL`. Actual plans unknown.
+
+Candidate indexes:
+
+```sql
+-- For ① queue_count: enables index-only COUNT
+CREATE INDEX idx_msg_queues_live ON msg_queues ((1)) WHERE deleted_at IS NULL;
+
+-- For ② notifier_count: enables index-only COUNT
+CREATE INDEX idx_msg_queues_live_notifier ON msg_queues ((1)) WHERE deleted_at IS NULL AND notifier_id IS NOT NULL;
+```
+
+**Trade-off**: Each index adds write overhead on every INSERT/UPDATE/DELETE touching the
+filtered columns. Need `EXPLAIN ANALYZE` to confirm the COUNT actually uses the index
+(PostgreSQL may choose seq scan if the partial index covers most rows).
+
+---
+
+## Not fixable (architectural)
+
+- **write_message / try_del_peek_msg max times (490-523ms)**: Lock contention on
+  `FOR UPDATE` of the same `recipient_id` row. Inherent to concurrent queue access — cannot
+  use `SKIP LOCKED` because these operations require the lock for correctness.
+
+- **UPDATE msg_queues SET updated_at (~430K calls, 83-90s total, 0.20ms avg)**: Per-call cost
+  is already minimal. Trigger does zero aggregate work for this pattern (verified — all
+  IS DISTINCT FROM checks fail, no `update_aggregates` called). Fix #3 eliminates even
+  the function call overhead.
+
+---
+
+## Summary
+
+| # | Fix | Per-call savings | Calls | Total savings | Verified |
+|---|-----|-----------------|-------|---------------|----------|
+| 1 | getEntityCounts ③④ → `SUM(queue_count)` | 0–158ms (unknown split across 4 subqueries) | ~2,500 | 0–395s | Correctness: yes. Savings: needs EXPLAIN ANALYZE |
+| 2 | Remove trailing COUNTs from expire_old_messages | 557–833ms (CSV-verified) | 10 | 5.6–8.3s | Yes (CSV verified) |
+| 3 | Add WHEN clause to tr_queue_update | ~0.02–0.05ms (PL/pgSQL entry overhead estimate) | ~1.1M skipped | ~22–55s | Correctness: yes. Savings: estimated, not measured |
+| 4 | Partial indexes for ①② | Unknown | ~2,500 | Unknown | No — needs EXPLAIN ANALYZE |

simplexmq.cabal

@@ -432,36 +432,6 @@ executable smp-server
     , text
   default-language: Haskell2010
 
-executable smp-server-bench
-  if flag(client_library)
-      buildable: False
-  if flag(server_postgres)
-    cpp-options: -DdbServerPostgres
-  main-is: Main.hs
-  other-modules:
-      ClientSim
-      Report
-  hs-source-dirs:
-      bench
-  default-extensions:
-      StrictData
-  ghc-options: -O2 -threaded -rtsopts
-  build-depends:
-      base
-    , async
-    , bytestring
-    , containers
-    , crypton
-    , mtl
-    , network
-    , simple-logger
-    , simplexmq
-    , stm
-    , text
-    , time
-    , unliftio
-  default-language: Haskell2010
-
 executable xftp
   if flag(client_library)
       buildable: False

src/Simplex/Messaging/Agent/Client.hs

@@ -1710,7 +1710,7 @@ processRcvServiceAssocs :: SMPQueue q => AgentClient -> [q] -> AM' ()
 processRcvServiceAssocs _ [] = pure ()
 processRcvServiceAssocs c serviceQs =
   withStore' c (`setRcvServiceAssocs` serviceQs) `catchAllErrors'` \e -> do
-    logError $ "processClientNotices error: " <> tshow e
+    logError $ "processRcvServiceAssocs error: " <> tshow e
     notifySub' c "" $ ERR e
 
 processClientNotices :: AgentClient -> SMPTransportSession -> [(RcvQueueSub, Maybe ClientNotice)] -> AM' ()

src/Simplex/Messaging/Server.hs

@@ -91,7 +91,7 @@ import qualified Data.X509 as X
 import qualified Data.X509.Validation as XV
 import GHC.Conc.Signal
 import GHC.IORef (atomicSwapIORef)
-import GHC.Stats (RTSStats (..), GCDetails (..), getRTSStats)
+import GHC.Stats (getRTSStats)
 import GHC.TypeLits (KnownNat)
 import Network.Socket (ServiceName, Socket, socketToHandle)
 import qualified Network.TLS as TLS
@@ -198,7 +198,6 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
             <> serverStatsThread_ cfg
             <> prometheusMetricsThread_ cfg
             <> controlPortThread_ cfg
-            <> [memoryDiagThread]
     )
     `finally` stopServer s
   where
@@ -720,75 +719,6 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
                 Nothing -> acc
                 Just (_, ts) -> (cnt + 1, updateTimeBuckets ts ts' times)
 
-    memoryDiagThread :: M s ()
-    memoryDiagThread = do
-      labelMyThread "memoryDiag"
-      Env
-        { ntfStore = NtfStore ntfMap,
-          server = srv@Server {subscribers, ntfSubscribers},
-          proxyAgent = ProxyAgent {smpAgent = pa},
-          msgStore_ = ms
-        } <- ask
-      let SMPClientAgent {smpClients, smpSessions, activeServiceSubs, activeQueueSubs, pendingServiceSubs, pendingQueueSubs, smpSubWorkers} = pa
-      liftIO $ forever $ do
-        threadDelay 300_000_000 -- 5 minutes
-        rts <- getRTSStats
-        let GCDetails {gcdetails_live_bytes, gcdetails_mem_in_use_bytes, gcdetails_large_objects_bytes, gcdetails_compact_bytes, gcdetails_block_fragmentation_bytes} = gc rts
-        clientCount <- IM.size <$> getServerClients srv
-        smpQSubs <- M.size <$> getSubscribedClients (queueSubscribers subscribers)
-        smpSSubs <- M.size <$> getSubscribedClients (serviceSubscribers subscribers)
-        ntfQSubs <- M.size <$> getSubscribedClients (queueSubscribers ntfSubscribers)
-        ntfSSubs <- M.size <$> getSubscribedClients (serviceSubscribers ntfSubscribers)
-        smpPending <- IM.size <$> readTVarIO (pendingEvents subscribers)
-        ntfPending <- IM.size <$> readTVarIO (pendingEvents ntfSubscribers)
-        ntfStoreSize <- M.size <$> readTVarIO ntfMap
-        paClients' <- M.size <$> readTVarIO smpClients
-        paSessions' <- M.size <$> readTVarIO smpSessions
-        paActSvc <- M.size <$> readTVarIO activeServiceSubs
-        paActQ <- M.size <$> readTVarIO activeQueueSubs
-        paPndSvc <- M.size <$> readTVarIO pendingServiceSubs
-        paPndQ <- M.size <$> readTVarIO pendingQueueSubs
-        paWorkers <- M.size <$> readTVarIO smpSubWorkers
-        lc <- loadedQueueCounts $ fromMsgStore ms
-        -- per-client metrics: total subscriptions and queue fill
-        clients <- getServerClients srv
-        let clientsList = IM.elems clients
-        totalSubs <- sum <$> mapM (\Client {subscriptions} -> M.size <$> readTVarIO subscriptions) clientsList
-        totalSndQ <- sum <$> mapM (\Client {sndQ} -> fromIntegral <$> atomically (lengthTBQueue sndQ)) clientsList
-        totalMsgQ <- sum <$> mapM (\Client {msgQ} -> fromIntegral <$> atomically (lengthTBQueue msgQ)) clientsList
-        totalEndThreads <- sum <$> mapM (\Client {endThreads} -> IM.size <$> readTVarIO endThreads) clientsList
-        logInfo $
-          "MEMORY"
-            <> " rts_live=" <> tshow gcdetails_live_bytes
-            <> " rts_heap=" <> tshow gcdetails_mem_in_use_bytes
-            <> " rts_max_live=" <> tshow (max_live_bytes rts)
-            <> " rts_large=" <> tshow gcdetails_large_objects_bytes
-            <> " rts_compact=" <> tshow gcdetails_compact_bytes
-            <> " rts_frag=" <> tshow gcdetails_block_fragmentation_bytes
-            <> " rts_gc=" <> tshow (gcs rts)
-            <> " clients=" <> tshow clientCount
-            <> " clientSubs=" <> tshow totalSubs
-            <> " clientSndQ=" <> tshow (totalSndQ :: Int)
-            <> " clientMsgQ=" <> tshow (totalMsgQ :: Int)
-            <> " clientThreads=" <> tshow totalEndThreads
-            <> " smpQSubs=" <> tshow smpQSubs
-            <> " smpSSubs=" <> tshow smpSSubs
-            <> " ntfQSubs=" <> tshow ntfQSubs
-            <> " ntfSSubs=" <> tshow ntfSSubs
-            <> " smpPending=" <> tshow smpPending
-            <> " ntfPending=" <> tshow ntfPending
-            <> " ntfStore=" <> tshow ntfStoreSize
-            <> " paClients=" <> tshow paClients'
-            <> " paSessions=" <> tshow paSessions'
-            <> " paActSvc=" <> tshow paActSvc
-            <> " paActQ=" <> tshow paActQ
-            <> " paPndSvc=" <> tshow paPndSvc
-            <> " paPndQ=" <> tshow paPndQ
-            <> " paWorkers=" <> tshow paWorkers
-            <> " loadedQ=" <> tshow (loadedQueueCount lc)
-            <> " loadedNtf=" <> tshow (loadedNotifierCount lc)
-            <> " ntfLocks=" <> tshow (notifierLockCount lc)
-
     runClient :: Transport c => X.CertificateChain -> C.APrivateSignKey -> TProxy c 'TServer -> c 'TServer -> M s ()
     runClient srvCert srvSignKey tp h = do
       ms <- asks msgStore

src/Simplex/Messaging/Server/Prometheus.hs

@@ -391,13 +391,13 @@ prometheusMetrics sm rtm ts =
       \# TYPE simplex_smp_ntf_services_queues_count gauge\n\
       \simplex_smp_ntf_services_queues_count " <> mshow (ntfServiceQueuesCount entityCounts) <> "\n# ntfServiceQueuesCount\n\
       \\n\
-      \# HELP simplex_smp_rcv_services_sub_msg The count of subscribed service queues with messages.\n\
-      \# TYPE simplex_smp_rcv_services_sub_msg counter\n\
-      \simplex_smp_rcv_services_sub_msg " <> mshow _rcvServicesSubMsg <> "\n# rcvServicesSubMsg\n\
+      \# HELP simplex_smp_rcv_services_sub_msg_count The count of subscribed service queues with messages.\n\
+      \# TYPE simplex_smp_rcv_services_sub_msg_count counter\n\
+      \simplex_smp_rcv_services_sub_msg_count " <> mshow _rcvServicesSubMsg <> "\n# rcvServicesSubMsg\n\
       \\n\
-      \# HELP simplex_smp_rcv_services_sub_duplicate The count of duplicate subscribed service queues.\n\
-      \# TYPE simplex_smp_rcv_services_sub_duplicate counter\n\
-      \simplex_smp_rcv_services_sub_duplicate " <> mshow _rcvServicesSubDuplicate <> "\n# rcvServicesSubDuplicate\n\
+      \# HELP simplex_smp_rcv_services_sub_duplicate_count The count of duplicate subscribed service queues.\n\
+      \# TYPE simplex_smp_rcv_services_sub_duplicate_count counter\n\
+      \simplex_smp_rcv_services_sub_duplicate_count " <> mshow _rcvServicesSubDuplicate <> "\n# rcvServicesSubDuplicate\n\
       \\n"
         <> showServices _rcvServices "rcv" "receiving"
         <> showServices _ntfServices "ntf" "notification"

tests/AgentTests/FunctionalAPITests.hs

@@ -494,6 +494,7 @@ functionalAPITests ps = do
     it "should re-subscribe when service ID changed" $ testClientServiceIDChange ps
     it "should clear pending service sub when service unavailable" $ testServiceUnavailableClearsPending ps
     it "should recover when service ID changes on reconnect" $ testServiceIdChangeOnReconnect ps
+    it "should handle service unavailable on startup" $ testServiceUnavailableOnStartup ps
     it "migrate connections to and from service" $ testMigrateConnectionsToService ps
   describe "Connection switch" $ do
     describe "should switch delivery to the new queue" $
@@ -3997,6 +3998,42 @@ testServiceIdChangeOnReconnect ps@(_, ASType qs _) = do
       ("", "", UP _ [_]) <- nGet user
       pure ()
 
+-- | Test that subscribeAllConnections handles service unavailable on startup.
+-- Agent has service credentials but server doesn't support services (askClientCert = False).
+testServiceUnavailableOnStartup :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testServiceUnavailableOnStartup (t, msType) = do
+  let srv = initAgentServersClientService
+      noSrv = initAgentServers
+  -- Phase 1: Establish connection with service
+  (sId, uId) <- withAgentClientsServers2 (agentCfg, srv) (agentCfg, noSrv) $ \service user ->
+    withSmpServerStoreLogOn (t, msType) testPort $ \_ -> runRight $ do
+      conns@(sId, uId) <- makeConnection service user
+      exchangeGreetings service uId user sId
+      pure conns
+  -- Phase 2: Server without service support, new service agent
+  let cfgNoService = updateCfg (cfgMS msType) $ \(cfg' :: ServerConfig s) ->
+        let ServerConfig {transportConfig} = cfg'
+         in cfg' {transportConfig = transportConfig {askClientCert = False}} :: ServerConfig s
+  -- Phase 2: Server without service support, service agent gets NO_SERVICE
+  withAgentClientsServers2 (agentCfg, srv) (agentCfg, noSrv) $ \service user ->
+    withSmpServerConfigOn t cfgNoService testPort $ \_ -> runRight $ do
+      subscribeAllConnections service False Nothing
+      ("", "", ERR (BROKER _ NO_SERVICE)) <- get service
+      ("", "", UP _ [_]) <- nGet service
+      subscribeAllConnections user False Nothing
+      ("", "", UP _ [_]) <- nGet user
+      exchangeGreetingsMsgId 4 service uId user sId
+  -- Phase 3: Normal server - cert was deleted, new cert generated,
+  -- no service sub in DB yet, queues subscribed individually
+  withAgentClientsServers2 (agentCfg, srv) (agentCfg, noSrv) $ \service user ->
+    withSmpServerStoreLogOn (t, msType) testPort $ \_ -> runRight $ do
+      liftIO $ threadDelay 250000
+      subscribeAllConnections service False Nothing
+      ("", "", UP _ [_]) <- nGet service
+      subscribeAllConnections user False Nothing
+      ("", "", UP _ [_]) <- nGet user
+      exchangeGreetingsMsgId 6 service uId user sId
+
 testMigrateConnectionsToService :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
 testMigrateConnectionsToService ps = do
   (((sId1, uId1), (uId2, sId2)), ((sId3, uId3), (uId4, sId4)), ((sId5, uId5), (uId6, sId6))) <-

tests/Test.hs

@@ -42,6 +42,7 @@ import XFTPWebTests (xftpWebTests)
 import Fixtures
 import SMPAgentClient (testDB)
 import Simplex.Messaging.Agent.Store.Postgres.Migrations.App
+import Simplex.Messaging.Agent.Store.Postgres.Util (dropAllSchemasExceptSystem)
 #else
 import AgentTests.SchemaDump (schemaDumpTest)
 #endif
@@ -149,7 +150,11 @@ main = do
           describe "XFTP file description" fileDescriptionTests
           describe "XFTP CLI" xftpCLITests
           describe "XFTP agent" xftpAgentTests
-        describe "XFTP Web Client" xftpWebTests
+#if defined(dbPostgres)
+        describe "XFTP Web Client" $ xftpWebTests (dropAllSchemasExceptSystem testDBConnectInfo)
+#else
+        describe "XFTP Web Client" $ xftpWebTests (pure ())
+#endif
         describe "XRCP" remoteControlTests
         describe "Web" webTests
         describe "Server CLIs" cliTests

tests/XFTPWebTests.hs

@@ -50,7 +50,7 @@ import AgentTests.FunctionalAPITests (rfGet, runRight, runRight_, sfGet, withAge
 import Simplex.Messaging.Agent (AgentClient, xftpReceiveFile, xftpSendFile, xftpStartWorkers)
 import Simplex.Messaging.Agent.Protocol (AEvent (..))
 import SMPAgentClient (agentCfg, initAgentServers, testDB)
-import XFTPCLI (recipientFiles, senderFiles)
+import XFTPCLI (recipientFiles, senderFiles, testBracket)
 import qualified Simplex.Messaging.Crypto.File as CF
 
 xftpWebDir :: FilePath
@@ -167,8 +167,8 @@ impAddr = "import * as Addr from './dist/protocol/address.js';"
 jsOut :: String -> String
 jsOut expr = "process.stdout.write(Buffer.from(" <> expr <> "));"
 
-xftpWebTests :: Spec
-xftpWebTests = do
+xftpWebTests :: IO () -> Spec
+xftpWebTests dbCleanup = do
   distExists <- runIO $ doesDirectoryExist (xftpWebDir <> "/dist")
   if distExists
     then do
@@ -187,7 +187,7 @@ xftpWebTests = do
       tsClientTests
       tsDownloadTests
       tsAddressTests
-      tsIntegrationTests
+      tsIntegrationTests dbCleanup
     else
       it "skipped (run 'cd xftp-web && npm install && npm run build' first)" $
         pendingWith "TS project not compiled"
@@ -2829,8 +2829,9 @@ tsAddressTests = describe "protocol/address" $ do
 
 -- ── integration ───────────────────────────────────────────────────
 
-tsIntegrationTests :: Spec
-tsIntegrationTests = describe "integration" $ do
+tsIntegrationTests :: IO () -> Spec
+tsIntegrationTests dbCleanup = describe "integration" $
+  around_ testBracket . after_ dbCleanup $ do
   it "web handshake with Ed25519 identity verification" $
     webHandshakeTest testXFTPServerConfigEd25519SNI "tests/fixtures/ed25519/ca.crt"
   it "web handshake with Ed448 identity verification" $