diff --git a/src/Simplex/Messaging/Server/MsgStore/Postgres.hs b/src/Simplex/Messaging/Server/MsgStore/Postgres.hs index 14dd3977a..721a71946 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Postgres.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Postgres.hs @@ -109,9 +109,12 @@ 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)" (now, ttl, 10000 :: Int) + DB.query db "CALL expire_old_messages(?,?,?,0,0,0)" (oldQueue, oldMsg, batchSize) where st = dbStore $ queueStore_ ms + oldQueue = 0 :: Int64 -- expire all queues + oldMsg = now - ttl + batchSize = 10000 :: Int toMessageStats (expiredMsgsCount, storedMsgsCount, storedQueues) = MessageStats {expiredMsgsCount, storedMsgsCount, storedQueues} @@ -298,8 +301,8 @@ deleteAllMessages ms = db [sql| UPDATE msg_queues - SET msg_queue_size = 0, msg_can_write = TRUE - WHERE msg_queue_size != 0 OR msg_can_write = FALSE + SET msg_queue_size = 0, msg_can_write = TRUE, msg_queue_expire = FALSE + WHERE msg_queue_size != 0 OR msg_can_write = FALSE OR msg_queue_expire = TRUE |] updateQueueCounts :: PostgresMsgStore -> IO () @@ -311,7 +314,7 @@ updateQueueCounts ms = CREATE TEMP TABLE queue_stats AS SELECT recipient_id, COUNT(*) AS size, - BOOL_OR(msg_quota) AS has_quota + SUM(CASE WHEN msg_quota THEN 1 ELSE 0 END) AS quota_count FROM messages GROUP BY recipient_id |] @@ -319,15 +322,16 @@ updateQueueCounts ms = db [sql| UPDATE msg_queues - SET msg_queue_size = 0, msg_can_write = TRUE - WHERE msg_queue_size != 0 OR msg_can_write = FALSE + SET msg_queue_size = 0, msg_can_write = TRUE, msg_queue_expire = FALSE + WHERE msg_queue_size != 0 OR msg_can_write = FALSE OR msg_queue_expire = TRUE |] void $ DB.execute_ db [sql| UPDATE msg_queues q SET msg_queue_size = s.size, - msg_can_write = NOT s.has_quota + msg_can_write = s.quota_count = 0, + msg_queue_expire = s.size > s.quota_count FROM queue_stats s WHERE q.recipient_id = s.recipient_id |] diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs b/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs index 48f22cf81..be14202c6 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs +++ b/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs @@ -177,6 +177,7 @@ CREATE TABLE messages( ALTER TABLE msg_queues ADD COLUMN msg_can_write BOOLEAN NOT NULL DEFAULT TRUE, + ADD COLUMN msg_queue_expire BOOLEAN NOT NULL DEFAULT FALSE, ADD COLUMN msg_queue_size BIGINT NOT NULL DEFAULT 0; CREATE INDEX idx_messages_recipient_id_message_id ON messages (recipient_id, message_id); @@ -184,7 +185,7 @@ CREATE INDEX idx_messages_recipient_id_msg_ts on messages(recipient_id, msg_ts); CREATE INDEX idx_messages_recipient_id_msg_quota on messages(recipient_id, msg_quota); DROP INDEX idx_msg_queues_updated_at; -CREATE INDEX idx_msg_queues_updated_at_recipient_id ON msg_queues (deleted_at, updated_at, msg_queue_size, recipient_id); +CREATE INDEX idx_msg_queues_updated_at_recipient_id ON msg_queues (deleted_at, updated_at, msg_queue_expire, recipient_id); CREATE FUNCTION write_message( p_recipient_id BYTEA, @@ -215,6 +216,7 @@ BEGIN UPDATE msg_queues SET msg_can_write = NOT quota_written, + msg_queue_expire = TRUE, msg_queue_size = msg_queue_size + 1 WHERE recipient_id = p_recipient_id; @@ -248,7 +250,9 @@ BEGIN IF NOT FOUND THEN IF q_size != 0 THEN UPDATE msg_queues - SET msg_can_write = TRUE, msg_queue_size = 0 + SET msg_can_write = TRUE, + msg_queue_expire = FALSE, + msg_queue_size = 0 WHERE recipient_id = p_recipient_id; END IF; RETURN; @@ -259,6 +263,7 @@ BEGIN IF FOUND THEN UPDATE msg_queues SET msg_can_write = msg_can_write OR msg_queue_size <= 1, + msg_queue_expire = msg_queue_size > 1, msg_queue_size = GREATEST(msg_queue_size - 1, 0) WHERE recipient_id = p_recipient_id; RETURN QUERY VALUES (msg.msg_id, msg.msg_ts, msg.msg_quota, msg.msg_ntf_flag, msg.msg_body); @@ -293,7 +298,9 @@ BEGIN IF NOT FOUND THEN IF q_size != 0 THEN UPDATE msg_queues - SET msg_can_write = TRUE, msg_queue_size = 0 + SET msg_can_write = TRUE, + msg_queue_expire = FALSE, + msg_queue_size = 0 WHERE recipient_id = p_recipient_id; END IF; RETURN; @@ -318,12 +325,15 @@ BEGIN IF msg_deleted THEN UPDATE msg_queues SET msg_can_write = msg_can_write OR msg_queue_size <= 1, + msg_queue_expire = msg_queue_size > 1, msg_queue_size = GREATEST(msg_queue_size - 1, 0) WHERE recipient_id = p_recipient_id; END IF; ELSIF msg_deleted OR q_size != 0 THEN UPDATE msg_queues - SET msg_can_write = TRUE, msg_queue_size = 0 + SET msg_can_write = TRUE, + msg_queue_expire = FALSE, + msg_queue_size = 0 WHERE recipient_id = p_recipient_id; END IF; ELSE @@ -336,7 +346,7 @@ CREATE FUNCTION delete_expired_msgs(p_recipient_id BYTEA, p_old_ts BIGINT) RETUR LANGUAGE plpgsql AS $$ DECLARE q_size BIGINT; - min_id BIGINT; + keep_min_id BIGINT; del_count BIGINT; BEGIN SELECT msg_queue_size INTO q_size @@ -348,21 +358,20 @@ BEGIN RETURN 0; END IF; - SELECT LEAST( -- ignores NULLs - (SELECT MIN(message_id) FROM messages WHERE recipient_id = p_recipient_id AND msg_ts >= p_old_ts), - (SELECT MIN(message_id) FROM messages WHERE recipient_id = p_recipient_id AND msg_quota = TRUE) - ) INTO min_id; + SELECT MIN(message_id) INTO keep_min_id + FROM messages WHERE recipient_id = p_recipient_id AND msg_ts >= p_old_ts AND msg_quota = FALSE; - IF min_id IS NULL THEN - DELETE FROM messages WHERE recipient_id = p_recipient_id; + IF keep_min_id IS NULL THEN + DELETE FROM messages WHERE recipient_id = p_recipient_id AND msg_quota = FALSE; ELSE - DELETE FROM messages WHERE recipient_id = p_recipient_id AND message_id < min_id; + DELETE FROM messages WHERE recipient_id = p_recipient_id AND message_id < keep_min_id AND msg_quota = FALSE; END IF; GET DIAGNOSTICS del_count = ROW_COUNT; IF del_count > 0 THEN UPDATE msg_queues SET msg_can_write = msg_can_write OR msg_queue_size <= del_count, + msg_queue_expire = msg_queue_size > del_count AND keep_min_id IS NOT NULL, msg_queue_size = GREATEST(msg_queue_size - del_count, 0) WHERE recipient_id = p_recipient_id; END IF; @@ -371,8 +380,8 @@ END; $$; CREATE PROCEDURE expire_old_messages( - p_now_ts BIGINT, - p_ttl BIGINT, + p_old_queue BIGINT, + p_old_ts BIGINT, batch_size INT, OUT r_expired_msgs_count BIGINT, OUT r_stored_msgs_count BIGINT, @@ -380,8 +389,6 @@ CREATE PROCEDURE expire_old_messages( ) LANGUAGE plpgsql AS $$ DECLARE - old_ts BIGINT := p_now_ts - p_ttl; - very_old_ts BIGINT := p_now_ts - 2 * p_ttl - 86400; rids BYTEA[]; rid BYTEA; last_rid BYTEA := '\x'; @@ -395,8 +402,8 @@ BEGIN SELECT recipient_id FROM msg_queues WHERE deleted_at IS NULL - AND updated_at > very_old_ts - AND msg_queue_size > 0 + AND updated_at > p_old_queue + AND msg_queue_expire = TRUE AND recipient_id > last_rid ORDER BY recipient_id ASC LIMIT batch_size @@ -407,7 +414,7 @@ BEGIN FOREACH rid IN ARRAY rids LOOP BEGIN - del_count := delete_expired_msgs(rid, old_ts); + 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; @@ -444,6 +451,7 @@ DROP INDEX idx_messages_recipient_id_msg_quota; ALTER TABLE msg_queues DROP COLUMN msg_can_write, + DROP COLUMN msg_queue_expire, DROP COLUMN msg_queue_size; DROP TABLE messages; diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql b/src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql index 5a98edb0d..433d45473 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql +++ b/src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql @@ -21,7 +21,7 @@ CREATE FUNCTION smp_server.delete_expired_msgs(p_recipient_id bytea, p_old_ts bi AS $$ DECLARE q_size BIGINT; - min_id BIGINT; + keep_min_id BIGINT; del_count BIGINT; BEGIN SELECT msg_queue_size INTO q_size @@ -33,21 +33,20 @@ BEGIN RETURN 0; END IF; - SELECT LEAST( -- ignores NULLs - (SELECT MIN(message_id) FROM messages WHERE recipient_id = p_recipient_id AND msg_ts >= p_old_ts), - (SELECT MIN(message_id) FROM messages WHERE recipient_id = p_recipient_id AND msg_quota = TRUE) - ) INTO min_id; + SELECT MIN(message_id) INTO keep_min_id + FROM messages WHERE recipient_id = p_recipient_id AND msg_ts >= p_old_ts AND msg_quota = FALSE; - IF min_id IS NULL THEN - DELETE FROM messages WHERE recipient_id = p_recipient_id; + IF keep_min_id IS NULL THEN + DELETE FROM messages WHERE recipient_id = p_recipient_id AND msg_quota = FALSE; ELSE - DELETE FROM messages WHERE recipient_id = p_recipient_id AND message_id < min_id; + DELETE FROM messages WHERE recipient_id = p_recipient_id AND message_id < keep_min_id AND msg_quota = FALSE; END IF; GET DIAGNOSTICS del_count = ROW_COUNT; IF del_count > 0 THEN UPDATE msg_queues SET msg_can_write = msg_can_write OR msg_queue_size <= del_count, + msg_queue_expire = msg_queue_size > del_count AND keep_min_id IS NOT NULL, msg_queue_size = GREATEST(msg_queue_size - del_count, 0) WHERE recipient_id = p_recipient_id; END IF; @@ -57,12 +56,10 @@ $$; -CREATE PROCEDURE smp_server.expire_old_messages(IN p_now_ts bigint, IN p_ttl 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_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 - old_ts BIGINT := p_now_ts - p_ttl; - very_old_ts BIGINT := p_now_ts - 2 * p_ttl - 86400; rids BYTEA[]; rid BYTEA; last_rid BYTEA := '\x'; @@ -76,8 +73,8 @@ BEGIN SELECT recipient_id FROM msg_queues WHERE deleted_at IS NULL - AND updated_at > very_old_ts - AND msg_queue_size > 0 + AND updated_at > p_old_queue + AND msg_queue_expire = TRUE AND recipient_id > last_rid ORDER BY recipient_id ASC LIMIT batch_size @@ -88,7 +85,7 @@ BEGIN FOREACH rid IN ARRAY rids LOOP BEGIN - del_count := delete_expired_msgs(rid, old_ts); + 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; @@ -132,7 +129,9 @@ BEGIN IF NOT FOUND THEN IF q_size != 0 THEN UPDATE msg_queues - SET msg_can_write = TRUE, msg_queue_size = 0 + SET msg_can_write = TRUE, + msg_queue_expire = FALSE, + msg_queue_size = 0 WHERE recipient_id = p_recipient_id; END IF; RETURN; @@ -143,6 +142,7 @@ BEGIN IF FOUND THEN UPDATE msg_queues SET msg_can_write = msg_can_write OR msg_queue_size <= 1, + msg_queue_expire = msg_queue_size > 1, msg_queue_size = GREATEST(msg_queue_size - 1, 0) WHERE recipient_id = p_recipient_id; RETURN QUERY VALUES (msg.msg_id, msg.msg_ts, msg.msg_quota, msg.msg_ntf_flag, msg.msg_body); @@ -179,7 +179,9 @@ BEGIN IF NOT FOUND THEN IF q_size != 0 THEN UPDATE msg_queues - SET msg_can_write = TRUE, msg_queue_size = 0 + SET msg_can_write = TRUE, + msg_queue_expire = FALSE, + msg_queue_size = 0 WHERE recipient_id = p_recipient_id; END IF; RETURN; @@ -204,12 +206,15 @@ BEGIN IF msg_deleted THEN UPDATE msg_queues SET msg_can_write = msg_can_write OR msg_queue_size <= 1, + msg_queue_expire = msg_queue_size > 1, msg_queue_size = GREATEST(msg_queue_size - 1, 0) WHERE recipient_id = p_recipient_id; END IF; ELSIF msg_deleted OR q_size != 0 THEN UPDATE msg_queues - SET msg_can_write = TRUE, msg_queue_size = 0 + SET msg_can_write = TRUE, + msg_queue_expire = FALSE, + msg_queue_size = 0 WHERE recipient_id = p_recipient_id; END IF; ELSE @@ -241,6 +246,7 @@ BEGIN UPDATE msg_queues SET msg_can_write = NOT quota_written, + msg_queue_expire = TRUE, msg_queue_size = msg_queue_size + 1 WHERE recipient_id = p_recipient_id; @@ -303,6 +309,7 @@ CREATE TABLE smp_server.msg_queues ( rcv_service_id bytea, ntf_service_id bytea, msg_can_write boolean DEFAULT true NOT NULL, + msg_queue_expire boolean DEFAULT false NOT NULL, msg_queue_size bigint DEFAULT 0 NOT NULL ); @@ -375,7 +382,7 @@ 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_size, recipient_id); +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); diff --git a/tests/CoreTests/MsgStoreTests.hs b/tests/CoreTests/MsgStoreTests.hs index ed5095fb4..3961a9ce0 100644 --- a/tests/CoreTests/MsgStoreTests.hs +++ b/tests/CoreTests/MsgStoreTests.hs @@ -327,34 +327,35 @@ testUpdateMessageCounts ms = do q <- ExceptT $ addQueue ms rId qr let write s = writeMsg ms q True =<< mkMessage s hasSize = checkQueueSize ms - q `hasSize` (0, True) + q `hasSize` (0, True, False) Just (Message {msgId = mId1}, True) <- write "message 1" - q `hasSize` (1, True) + q `hasSize` (1, True, True) Just (Message {msgId = mId2}, False) <- write "message 2" - q `hasSize` (2, True) + q `hasSize` (2, True, True) Just (Message {msgId = mId3}, False) <- write "message 3" - q `hasSize` (3, True) + q `hasSize` (3, True, True) Nothing <- write "message 4" - q `hasSize` (4, False) + q `hasSize` (4, False, True) Msg "message 1" <- tryPeekMsg ms q - q `hasSize` (4, False) + q `hasSize` (4, False, True) Msg "message 1" <- tryDelMsg ms q mId1 - q `hasSize` (3, False) + q `hasSize` (3, False, True) Msg "message 2" <- tryPeekMsg ms q (Msg "message 2", Msg "message 3") <- tryDelPeekMsg ms q mId2 - q `hasSize` (2, False) + q `hasSize` (2, False, True) (Msg "message 3", Just MessageQuota {msgId = mId4}) <- tryDelPeekMsg ms q mId3 - q `hasSize` (1, False) + q `hasSize` (1, False, True) (Just MessageQuota {}, Nothing) <- tryDelPeekMsg ms q mId4 - q `hasSize` (0, True) + q `hasSize` (0, True, False) -checkQueueSize :: PostgresMsgStore -> PostgresQueue -> (Int64, Bool) -> ExceptT ErrorType IO () -checkQueueSize ms q (size, canWrt) = liftIO $ do - [(size', canWrt')] <- +checkQueueSize :: PostgresMsgStore -> PostgresQueue -> (Int64, Bool, Bool) -> ExceptT ErrorType IO () +checkQueueSize ms q (size, canWrt, expire) = liftIO $ do + [(size', canWrt', expire')] <- withTransaction (dbStore $ queueStore ms) $ \db -> - DB.query db "SELECT msg_queue_size, msg_can_write FROM msg_queues WHERE recipient_id = ?" (Only (recipientId q)) + DB.query db "SELECT msg_queue_size, msg_can_write, msg_queue_expire FROM msg_queues WHERE recipient_id = ?" (Only (recipientId q)) size' `shouldBe` size canWrt' `shouldBe` canWrt + expire' `shouldBe` expire testResetMessageCounts :: PostgresMsgStore -> IO () testResetMessageCounts ms = do @@ -368,20 +369,20 @@ testResetMessageCounts ms = do Just (Message {msgId = mId2}, False) <- write "message 2" Just (Message {msgId = mId3}, False) <- write "message 3" Nothing <- write "message 4" - q `hasSize` (4, False) + q `hasSize` (4, False, True) liftIO $ setIncorrectSize q (10, True) Nothing <- write "message 5" - q `hasSize` (11, False) + q `hasSize` (11, False, True) (Msg "message 1", Msg "message 2") <- tryDelPeekMsg ms q mId1 - q `hasSize` (10, False) + q `hasSize` (10, False, True) (Msg "message 2", Msg "message 3") <- tryDelPeekMsg ms q mId2 - q `hasSize` (9, False) + q `hasSize` (9, False, True) (Msg "message 3", Just MessageQuota {msgId = mId4}) <- tryDelPeekMsg ms q mId3 - q `hasSize` (8, False) + q `hasSize` (8, False, True) (Just MessageQuota {}, Just MessageQuota {msgId = mId5}) <- tryDelPeekMsg ms q mId4 - q `hasSize` (7, False) + q `hasSize` (7, False, True) (Just MessageQuota {}, Nothing) <- tryDelPeekMsg ms q mId5 - q `hasSize` (0, True) -- reset + q `hasSize` (0, True, False) -- reset where setIncorrectSize :: PostgresQueue -> (Int64, Bool) -> IO () setIncorrectSize q (size, canWrt) =