diff --git a/plans/2026-10-08-fast-rotation-old-queue-drain.md b/plans/2026-10-08-fast-rotation-old-queue-drain.md index 8cf3413cf..aa226bf82 100644 --- a/plans/2026-10-08-fast-rotation-old-queue-drain.md +++ b/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. diff --git a/simplexmq.cabal b/simplexmq.cabal index ef46fb723..0153db450 100644 --- a/simplexmq.cabal +++ b/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: diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs index 26c7a80da..4e473f473 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs +++ b/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 diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20261008_rcv_switch_status.hs b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20261008_rcv_switch_status.hs new file mode 100644 index 000000000..5ec0ae377 --- /dev/null +++ b/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; +|] diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql index 55a9ab78d..ff28100f5 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql +++ b/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); diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/App.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/App.hs index 856c86626..5dbd314f9 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/App.hs +++ b/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 diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20261008_rcv_switch_status.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20261008_rcv_switch_status.hs new file mode 100644 index 000000000..cf0283d7e --- /dev/null +++ b/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; +|] diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql index 963f918ac..cdbcee7dd 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql +++ b/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 diff --git a/tests/AgentTests/SchemaDump.hs b/tests/AgentTests/SchemaDump.hs index f7435a23b..c64d45e1e 100644 --- a/tests/AgentTests/SchemaDump.hs +++ b/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