smp-server: reduce Prometheus and expiration DB load (#1872)

* smp server: add prometheus scan indexes

* smp server: estimate queue count in metrics

* docs: add smp server db load report

* smp server: reduce msg_queues write load

* docs: document batch-2 fixes and operator actions

src/Simplex/Messaging/Server/MsgStore/Postgres.hs

@@ -110,10 +110,9 @@ instance MsgStoreClass PostgresMsgStore where
   expireOldMessages :: Bool -> PostgresMsgStore -> Int64 -> Int64 -> IO MessageStats
   expireOldMessages _tty ms now ttl =
     maybeFirstRow' newMessageStats toMessageStats $ withConnection st $ \db ->
-      DB.query db "CALL expire_old_messages(?,?,?,0,0,0)" (oldQueue, oldMsg, batchSize)
+      DB.query db "CALL expire_old_messages(?,?,0,0,0)" (oldMsg, batchSize)
     where
       st = dbStore $ queueStore_ ms
-      oldQueue = 0 :: Int64 -- expire all queues
       oldMsg = now - ttl
       batchSize = 10000 :: Int
       toMessageStats (expiredMsgsCount, storedMsgsCount, storedQueues) =

src/Simplex/Messaging/Server/QueueStore/Postgres.hs

@@ -159,7 +159,8 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
           db
           [sql|
             SELECT
-              (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL) AS queue_count,
+              -- estimate via reltuples to avoid a full heap scan on every scrape
+              (SELECT GREATEST(reltuples, 0)::bigint FROM pg_class WHERE oid = 'msg_queues'::regclass) 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,
               (SELECT COUNT(1) FROM services WHERE service_role = ?) AS ntf_service_count,

src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs

@@ -20,7 +20,9 @@ serverSchemaMigrations =
     ("20250320_short_links", m20250320_short_links, Just down_m20250320_short_links),
     ("20250514_service_certs", m20250514_service_certs, Just down_m20250514_service_certs),
     ("20250903_store_messages", m20250903_store_messages, Just down_m20250903_store_messages),
-    ("20250915_queue_ids_hash", m20250915_queue_ids_hash, Just down_m20250915_queue_ids_hash)
+    ("20250915_queue_ids_hash", m20250915_queue_ids_hash, Just down_m20250915_queue_ids_hash),
+    ("20260916_prometheus_indexes", m20260916_prometheus_indexes, Just down_m20260916_prometheus_indexes),
+    ("20260917_msg_queues_hot", m20260917_msg_queues_hot, Just down_m20260917_msg_queues_hot)
   ]
 
 -- | The list of migrations in ascending order by date
@@ -588,3 +590,147 @@ ALTER TABLE services
   DROP COLUMN queue_ids_hash;
     |]
     <> dropXorHashFuncs
+
+m20260916_prometheus_indexes :: Text
+m20260916_prometheus_indexes =
+  [r|
+CREATE INDEX idx_msg_queues_expire ON msg_queues (recipient_id) WHERE deleted_at IS NULL AND msg_queue_expire;
+CREATE INDEX idx_msg_queues_notifier_active ON msg_queues (notifier_id) WHERE deleted_at IS NULL AND notifier_id IS NOT NULL;
+    |]
+
+down_m20260916_prometheus_indexes :: Text
+down_m20260916_prometheus_indexes =
+  [r|
+DROP INDEX idx_msg_queues_expire;
+DROP INDEX idx_msg_queues_notifier_active;
+    |]
+
+m20260917_msg_queues_hot :: Text
+m20260917_msg_queues_hot =
+  [r|
+DROP INDEX idx_msg_queues_updated_at_recipient_id;
+
+ALTER TABLE msg_queues SET (
+  fillfactor = 80,
+  autovacuum_vacuum_scale_factor = 0.02,
+  autovacuum_analyze_scale_factor = 0.01,
+  autovacuum_vacuum_cost_limit = 1000
+);
+
+ALTER TABLE messages SET (
+  autovacuum_vacuum_scale_factor = 0.02,
+  autovacuum_analyze_scale_factor = 0.01,
+  toast.autovacuum_vacuum_scale_factor = 0.02
+);
+
+ALTER TABLE services SET (
+  fillfactor = 70,
+  autovacuum_vacuum_threshold = 1000,
+  autovacuum_vacuum_scale_factor = 0
+);
+
+DROP PROCEDURE expire_old_messages(bigint, bigint, integer);
+
+CREATE PROCEDURE expire_old_messages(IN p_old_ts bigint, IN batch_size integer, OUT r_expired_msgs_count bigint, OUT r_stored_msgs_count bigint, OUT r_stored_queues bigint)
+    LANGUAGE plpgsql
+    AS $$
+DECLARE
+  rids BYTEA[];
+  rid BYTEA;
+  last_rid BYTEA := '\x';
+  del_count BIGINT;
+  total_deleted BIGINT := 0;
+BEGIN
+  LOOP
+    SELECT array_agg(recipient_id)
+    INTO rids
+    FROM (
+      SELECT recipient_id
+      FROM msg_queues
+      WHERE deleted_at IS NULL
+        AND msg_queue_expire = TRUE
+        AND recipient_id > last_rid
+      ORDER BY recipient_id ASC
+      LIMIT batch_size
+    ) qs;
+
+    EXIT WHEN rids IS NULL OR cardinality(rids) = 0;
+
+    FOREACH rid IN ARRAY rids
+    LOOP
+      BEGIN
+        del_count := delete_expired_msgs(rid, p_old_ts);
+        total_deleted := total_deleted + del_count;
+      EXCEPTION WHEN OTHERS THEN
+        RAISE WARNING 'STORE, expire_old_messages, error expiring queue %: %', encode(rid, 'base64'), SQLERRM;
+        CONTINUE;
+      END;
+      COMMIT;
+    END LOOP;
+    last_rid := rids[cardinality(rids)];
+  END LOOP;
+
+  r_expired_msgs_count := total_deleted;
+  r_stored_msgs_count := (SELECT COUNT(1) FROM messages);
+  r_stored_queues := (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL);
+END;
+$$;
+    |]
+
+down_m20260917_msg_queues_hot :: Text
+down_m20260917_msg_queues_hot =
+  [r|
+DROP PROCEDURE expire_old_messages(bigint, integer);
+
+CREATE PROCEDURE expire_old_messages(IN p_old_queue bigint, IN p_old_ts bigint, IN batch_size integer, OUT r_expired_msgs_count bigint, OUT r_stored_msgs_count bigint, OUT r_stored_queues bigint)
+    LANGUAGE plpgsql
+    AS $$
+DECLARE
+  rids BYTEA[];
+  rid BYTEA;
+  last_rid BYTEA := '\x';
+  del_count BIGINT;
+  total_deleted BIGINT := 0;
+BEGIN
+  LOOP
+    SELECT array_agg(recipient_id)
+    INTO rids
+    FROM (
+      SELECT recipient_id
+      FROM msg_queues
+      WHERE deleted_at IS NULL
+        AND updated_at > p_old_queue
+        AND msg_queue_expire = TRUE
+        AND recipient_id > last_rid
+      ORDER BY recipient_id ASC
+      LIMIT batch_size
+    ) qs;
+
+    EXIT WHEN rids IS NULL OR cardinality(rids) = 0;
+
+    FOREACH rid IN ARRAY rids
+    LOOP
+      BEGIN
+        del_count := delete_expired_msgs(rid, p_old_ts);
+        total_deleted := total_deleted + del_count;
+      EXCEPTION WHEN OTHERS THEN
+        RAISE WARNING 'STORE, expire_old_messages, error expiring queue %: %', encode(rid, 'base64'), SQLERRM;
+        CONTINUE;
+      END;
+      COMMIT;
+    END LOOP;
+    last_rid := rids[cardinality(rids)];
+  END LOOP;
+
+  r_expired_msgs_count := total_deleted;
+  r_stored_msgs_count := (SELECT COUNT(1) FROM messages);
+  r_stored_queues := (SELECT COUNT(1) FROM msg_queues WHERE deleted_at IS NULL);
+END;
+$$;
+
+CREATE INDEX idx_msg_queues_updated_at_recipient_id ON msg_queues (deleted_at, updated_at, msg_queue_expire, recipient_id);
+
+ALTER TABLE msg_queues RESET (fillfactor, autovacuum_vacuum_scale_factor, autovacuum_analyze_scale_factor, autovacuum_vacuum_cost_limit);
+ALTER TABLE messages RESET (autovacuum_vacuum_scale_factor, autovacuum_analyze_scale_factor, toast.autovacuum_vacuum_scale_factor);
+ALTER TABLE services RESET (fillfactor, autovacuum_vacuum_threshold, autovacuum_vacuum_scale_factor);
+    |]

src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql

@@ -1,5 +1,6 @@
 
 
+
 SET statement_timeout = 0;
 SET lock_timeout = 0;
 SET idle_in_transaction_session_timeout = 0;
@@ -56,7 +57,7 @@ $$;
 
 
 
-CREATE PROCEDURE smp_server.expire_old_messages(IN p_old_queue bigint, IN p_old_ts bigint, IN batch_size integer, OUT r_expired_msgs_count bigint, OUT r_stored_msgs_count bigint, OUT r_stored_queues bigint)
+CREATE PROCEDURE smp_server.expire_old_messages(IN p_old_ts bigint, IN batch_size integer, OUT r_expired_msgs_count bigint, OUT r_stored_msgs_count bigint, OUT r_stored_queues bigint)
     LANGUAGE plpgsql
     AS $$
 DECLARE
@@ -73,7 +74,6 @@ BEGIN
       SELECT recipient_id
       FROM msg_queues
       WHERE deleted_at IS NULL
-        AND updated_at > p_old_queue
         AND msg_queue_expire = TRUE
         AND recipient_id > last_rid
       ORDER BY recipient_id ASC
@@ -397,7 +397,8 @@ CREATE TABLE smp_server.messages (
     msg_quota boolean NOT NULL,
     msg_ntf_flag boolean NOT NULL,
     msg_body bytea NOT NULL
-);
+)
+WITH (autovacuum_vacuum_scale_factor='0.02', autovacuum_analyze_scale_factor='0.01', toast.autovacuum_vacuum_scale_factor='0.02');
 
 
 
@@ -441,7 +442,8 @@ CREATE TABLE smp_server.msg_queues (
     msg_can_write boolean DEFAULT true NOT NULL,
     msg_queue_expire boolean DEFAULT false NOT NULL,
     msg_queue_size bigint DEFAULT 0 NOT NULL
-);
+)
+WITH (fillfactor='80', autovacuum_vacuum_scale_factor='0.02', autovacuum_analyze_scale_factor='0.01', autovacuum_vacuum_cost_limit='1000');
 
 
 
@@ -453,7 +455,8 @@ CREATE TABLE smp_server.services (
     created_at bigint NOT NULL,
     queue_count bigint DEFAULT 0 NOT NULL,
     queue_ids_hash bytea DEFAULT '\x00000000000000000000000000000000'::bytea NOT NULL
-);
+)
+WITH (fillfactor='70', autovacuum_vacuum_threshold='1000', autovacuum_vacuum_scale_factor='0');
 
 
 
@@ -494,10 +497,18 @@ CREATE INDEX idx_messages_recipient_id_msg_ts ON smp_server.messages USING btree
 
 
 
+CREATE INDEX idx_msg_queues_expire ON smp_server.msg_queues USING btree (recipient_id) WHERE ((deleted_at IS NULL) AND msg_queue_expire);
+
+
+
 CREATE UNIQUE INDEX idx_msg_queues_link_id ON smp_server.msg_queues USING btree (link_id);
 
 
 
+CREATE INDEX idx_msg_queues_notifier_active ON smp_server.msg_queues USING btree (notifier_id) WHERE ((deleted_at IS NULL) AND (notifier_id IS NOT NULL));
+
+
+
 CREATE UNIQUE INDEX idx_msg_queues_notifier_id ON smp_server.msg_queues USING btree (notifier_id);
 
 
@@ -514,10 +525,6 @@ CREATE UNIQUE INDEX idx_msg_queues_sender_id ON smp_server.msg_queues USING btre
 
 
 
-CREATE INDEX idx_msg_queues_updated_at_recipient_id ON smp_server.msg_queues USING btree (deleted_at, updated_at, msg_queue_expire, recipient_id);
-
-
-
 CREATE INDEX idx_services_service_role ON smp_server.services USING btree (service_role);
 
 
@@ -549,3 +556,4 @@ ALTER TABLE ONLY smp_server.msg_queues
 
 
 
+