agent: migration to add index to read queues (#1934)

* agent: migration to add index to read queues

* update postgres schema

---------

Co-authored-by: Evgeny @ SimpleX Chat <[email protected]>

plans/2026-10-08-fast-rotation-old-queue-drain.md

@@ -27,6 +27,9 @@ Recipient only. The sender is unchanged.
   - `OK`: `True`.
   - `MSG`: `False`; the message is written to `msgQ`.
 - Apps: `received_qend` is added to `RcvSwitchStatus`.
+- Migration `20261008_rcv_switch_status`, SQLite and Postgres:
+  - up: index `idx_rcv_queues_switch_status` on `rcv_queues(switch_status)`.
+  - down: `received_qend` is set to `NULL`, the index is dropped.
 
 ## Mechanics
 
@@ -56,3 +59,4 @@ Recipient only. The sender is unchanged.
   - the old queue is deleted after the `ACK` that returns `OK`.
 - The same with a recipient restart before the `ACK`.
 - Old server stopped: the old queue is deleted by the bound.
+- Downgrade: `received_qend` is cleared, other switch statuses are kept.

simplexmq.cabal

@@ -196,6 +196,7 @@ library
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260823_snd_files_entitlement
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260919_ratchet_verify_codes
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260929_ratchet_indexes
+          Simplex.Messaging.Agent.Store.Postgres.Migrations.M20261008_rcv_switch_status
   else
       exposed-modules:
           Simplex.Messaging.Agent.Store.SQLite
@@ -253,6 +254,7 @@ library
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260919_ratchet_verify_codes
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260929_ratchet_indexes
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20261005_commands_autoincrement
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20261008_rcv_switch_status
           Simplex.Messaging.Agent.Store.SQLite.Util
   if flag(client_postgres) || flag(server_postgres)
       exposed-modules:

src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs

@@ -17,6 +17,7 @@ import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260712_address_dr_rp
 import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260823_snd_files_entitlement
 import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260919_ratchet_verify_codes
 import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20260929_ratchet_indexes
+import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20261008_rcv_switch_status
 import Simplex.Messaging.Agent.Store.Shared (Migration (..))
 
 schemaMigrations :: [(String, Text, Maybe Text)]
@@ -33,7 +34,8 @@ schemaMigrations =
     ("20260712_address_dr_rpc", m20260712_address_dr_rpc, Just down_m20260712_address_dr_rpc),
     ("20260823_snd_files_entitlement", m20260823_snd_files_entitlement, Just down_m20260823_snd_files_entitlement),
     ("20260919_ratchet_verify_codes", m20260919_ratchet_verify_codes, Just down_m20260919_ratchet_verify_codes),
-    ("20260929_ratchet_indexes", m20260929_ratchet_indexes, Just down_m20260929_ratchet_indexes)
+    ("20260929_ratchet_indexes", m20260929_ratchet_indexes, Just down_m20260929_ratchet_indexes),
+    ("20261008_rcv_switch_status", m20261008_rcv_switch_status, Just down_m20261008_rcv_switch_status)
   ]
 
 -- | The list of migrations in ascending order by date

src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20261008_rcv_switch_status.hs

@@ -0,0 +1,20 @@
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.Postgres.Migrations.M20261008_rcv_switch_status where
+
+import Data.Text (Text)
+import Text.RawString.QQ (r)
+
+m20261008_rcv_switch_status :: Text
+m20261008_rcv_switch_status =
+  [r|
+CREATE INDEX idx_rcv_queues_switch_status ON rcv_queues(switch_status);
+|]
+
+down_m20261008_rcv_switch_status :: Text
+down_m20261008_rcv_switch_status =
+  [r|
+UPDATE rcv_queues SET switch_status = NULL WHERE switch_status = 'received_qend';
+DROP INDEX idx_rcv_queues_switch_status;
+|]

src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql

@@ -1233,6 +1233,10 @@ CREATE UNIQUE INDEX idx_rcv_queues_ntf ON smp_agent_test_protocol_schema.rcv_que
 
 
 
+CREATE INDEX idx_rcv_queues_switch_status ON smp_agent_test_protocol_schema.rcv_queues USING btree (switch_status);
+
+
+
 CREATE INDEX idx_rcv_queues_to_subscribe ON smp_agent_test_protocol_schema.rcv_queues USING btree (to_subscribe);
 
 

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

@@ -54,6 +54,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260823_snd_files_entit
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260919_ratchet_verify_codes
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260929_ratchet_indexes
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20261005_commands_autoincrement
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20261008_rcv_switch_status
 import Simplex.Messaging.Agent.Store.Shared (Migration (..))
 
 schemaMigrations :: [(String, Query, Maybe Query)]
@@ -107,7 +108,8 @@ schemaMigrations =
     ("m20260823_snd_files_entitlement", m20260823_snd_files_entitlement, Just down_m20260823_snd_files_entitlement),
     ("m20260919_ratchet_verify_codes", m20260919_ratchet_verify_codes, Just down_m20260919_ratchet_verify_codes),
     ("m20260929_ratchet_indexes", m20260929_ratchet_indexes, Just down_m20260929_ratchet_indexes),
-    ("m20261005_commands_autoincrement", m20261005_commands_autoincrement, Just down_m20261005_commands_autoincrement)
+    ("m20261005_commands_autoincrement", m20261005_commands_autoincrement, Just down_m20261005_commands_autoincrement),
+    ("m20261008_rcv_switch_status", m20261008_rcv_switch_status, Just down_m20261008_rcv_switch_status)
   ]
 
 -- | The list of migrations in ascending order by date

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

@@ -0,0 +1,19 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20261008_rcv_switch_status where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20261008_rcv_switch_status :: Query
+m20261008_rcv_switch_status =
+  [sql|
+CREATE INDEX idx_rcv_queues_switch_status ON rcv_queues(switch_status);
+|]
+
+down_m20261008_rcv_switch_status :: Query
+down_m20261008_rcv_switch_status =
+  [sql|
+UPDATE rcv_queues SET switch_status = NULL WHERE switch_status = 'received_qend';
+DROP INDEX idx_rcv_queues_switch_status;
+|]

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql

@@ -645,6 +645,7 @@ CREATE INDEX idx_processed_ratchet_key_hashes_conn_id ON processed_ratchet_key_h
   conn_id,
   processed_ratchet_key_hash_id
 );
+CREATE INDEX idx_rcv_queues_switch_status ON rcv_queues(switch_status);
 CREATE TRIGGER tr_rcv_queue_insert
 AFTER INSERT ON rcv_queues
 FOR EACH ROW

tests/AgentTests/SchemaDump.hs

@@ -49,6 +49,7 @@ schemaDumpTest = do
   it "should NOT create user record for new database" testUsersMigrationNew
   it "should create user record for old database" testUsersMigrationOld
   it "should remove duplicate ratchet key hashes before adding unique index" testRatchetKeyHashesUniqueMigration
+  it "should clear received_qend switch status on downgrade" testRcvSwitchStatusDowngrade
 
 testVerifySchemaDump :: IO ()
 testVerifySchemaDump = do
@@ -129,6 +130,21 @@ testRatchetKeyHashesUniqueMigration = do
     `shouldReturn` [Only (1 :: Int), Only 3, Only 4]
   closeDBStore st'
 
+testRcvSwitchStatusDowngrade :: IO ()
+testRcvSwitchStatusDowngrade = do
+  Right st <- createDBStore (DBOpts testDB [] "" False True TQOff) appMigrations (MigrationConfig MCError Nothing)
+  withTransaction' st $ \db -> do
+    SQL.execute_ db "INSERT INTO users (user_id) VALUES (1)"
+    SQL.execute_ db "INSERT INTO connections (conn_id, conn_mode, user_id) VALUES (x'01', 'INV', 1)"
+    SQL.execute_ db "INSERT INTO servers (host, port, key_hash) VALUES ('localhost', '5001', x'00')"
+    SQL.execute_ db "INSERT INTO rcv_queues (host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, snd_id, status, rcv_queue_id, rcv_primary, switch_status) VALUES ('localhost', '5001', x'01', x'01', x'00', x'00', x'00', x'01', 'active', 1, 0, 'received_qend'), ('localhost', '5001', x'02', x'01', x'00', x'00', x'00', x'02', 'active', 2, 1, 'sending_qadd')"
+  closeDBStore st
+  let beforeRcvSwitchStatus = takeWhile (("m20261008_rcv_switch_status" /=) . name) appMigrations
+  Right st' <- createDBStore (DBOpts testDB [] "" False True TQOff) beforeRcvSwitchStatus (MigrationConfig MCYesUpDown Nothing)
+  withTransaction' st' (`SQL.query_` "SELECT switch_status FROM rcv_queues ORDER BY rcv_queue_id")
+    `shouldReturn` [Only (Nothing :: Maybe Text), Only (Just "sending_qadd")]
+  closeDBStore st'
+
 skipComparisonForDownMigrations :: [String]
 skipComparisonForDownMigrations =
   [ -- on down migration idx_messages_internal_snd_id_ts index moves down to the end of the file