From 5c3633477b19bd47accb2d3e258619d69ee58963 Mon Sep 17 00:00:00 2001 From: Kim Alvefur Date: Fri, 22 Mar 2019 17:32:56 +0100 Subject: mod_mam: Perform message expiry based on building an index by date (backport of 39ee70fbb009 from trunk) For each day, store a set of users that have new messages. To expire messages, we collect the union of sets of users from dates that fall outside the cleanup range. The previous algoritm did not work well with many users, especially with the default settings. --- plugins/mod_mam/mod_mam.lua | 74 ++++++++++++++++++++++++++------------------- 1 file changed, 43 insertions(+), 31 deletions(-) diff --git a/plugins/mod_mam/mod_mam.lua b/plugins/mod_mam/mod_mam.lua index 94bedbb1..a8c2689d 100644 --- a/plugins/mod_mam/mod_mam.lua +++ b/plugins/mod_mam/mod_mam.lua @@ -33,7 +33,7 @@ local is_stanza = st.is_stanza; local tostring = tostring; local time_now = os.time; local m_min = math.min; -local timestamp, timestamp_parse = require "util.datetime".datetime, require "util.datetime".parse; +local timestamp, timestamp_parse, datestamp = import( "util.datetime", "datetime", "parse", "date"); local default_max_items, max_max_items = 20, module:get_option_number("max_archive_query_results", 50); local strip_tags = module:get_option_set("dont_archive_namespaces", { "http://jabber.org/protocol/chatstates" }); @@ -46,13 +46,8 @@ if not archive.find then end local use_total = module:get_option_boolean("mam_include_total", true); -local cleanup; - -local function schedule_cleanup(username) - if cleanup and not cleanup[username] then - table.insert(cleanup, username); - cleanup[username] = true; - end +function schedule_cleanup() + -- replaced by non-noop later if cleanup is enabled end -- Handle prefs. @@ -96,7 +91,6 @@ module:hook("iq-set/self/"..xmlns_mam..":query", function(event) local qid = query.attr.queryid; get_prefs(origin.username, true); - schedule_cleanup(origin.username); -- Search query parameters local qwith, qstart, qend; @@ -212,6 +206,7 @@ end local function shall_store(user, who) -- TODO Cache this? if not um.user_exists(user, host) then + module:log("debug", "%s@%s does not exist", user, host) return false; end local prefs = get_prefs(user); @@ -329,6 +324,9 @@ module:hook("pre-message/full", strip_stanza_id_after_other_events, -1); local cleanup_after = module:get_option_string("archive_expires_after", "1w"); local cleanup_interval = module:get_option_number("archive_cleanup_interval", 4 * 60 * 60); if cleanup_after ~= "never" then + local cleanup_storage = module:open_store("archive_cleanup"); + local cleanup_map = module:open_store("archive_cleanup", "map"); + local day = 86400; local multipliers = { d = day, w = day * 7, m = 31 * day, y = 365.2425 * day }; local n, m = cleanup_after:lower():match("(%d+)%s*([dwmy]?)"); @@ -346,33 +344,47 @@ if cleanup_after ~= "never" then return false; end - -- Set of known users to do message expiry for - -- Populated either below or when new messages are added - cleanup = {}; + -- For each day, store a set of users that have new messages. To expire + -- messages, we collect the union of sets of users from dates that fall + -- outside the cleanup range. - -- Iterating over users is not supported by all authentication modules - -- Catch and ignore error if not supported - pcall(function () - -- If this works, then we schedule cleanup for all known users on startup - for user in um.users(module.host) do - schedule_cleanup(user); - end - end); + function schedule_cleanup(username, date) + cleanup_map:set(date or datestamp(), username, true); + end - -- At odd intervals, delete old messages for one user - module:add_timer(math.random(10, 60), function() - local user = table.remove(cleanup, 1); - if user then - module:log("debug", "Removing old messages for user %q", user); + cleanup_runner = require "util.async".runner(function () + local users = {}; + local cut_off = datestamp(os.time() - cleanup_after); + for date in cleanup_storage:users() do + if date <= cut_off then + module:log("debug", "Messages from %q should be expired", date); + local messages_this_day = cleanup_storage:get(date); + if messages_this_day then + for user in pairs(messages_this_day) do + users[user] = true; + end + if date < cut_off then + -- Messages from the same day as the cut-off might not have expired yet, + -- but all earlier will have, so clear storage for those days. + cleanup_storage:set(date, nil); + end + end + end + end + local sum, num_users = 0, 0; + for user in pairs(users) do local ok, err = archive:delete(user, { ["end"] = os.time() - cleanup_after; }) - if not ok then - module:log("warn", "Could not expire archives for user %s: %s", user, err); - elseif type(ok) == "number" then - module:log("debug", "Removed %d messages", ok); + if ok then + num_users = num_users + 1; + sum = sum + (tonumber(ok) or 0); end - cleanup[user] = nil; end - return math.random(cleanup_interval, cleanup_interval * 2); + module:log("info", "Deleted %d expired messages for %d users", sum, num_users); + end); + + cleanup_task = module:add_timer(1, function () + cleanup_runner:run(true); + return cleanup_interval; end); else module:log("debug", "Archive expiry disabled"); -- cgit v1.2.3 From 4c0bdc6b5c812883b802f3a9b02000a3a306b45d Mon Sep 17 00:00:00 2001 From: Kim Alvefur Date: Fri, 22 Mar 2019 17:40:07 +0100 Subject: mod_muc_mam: Copy cleanup mechanism from mod_mam (fixes #672) --- plugins/mod_muc_mam.lua | 79 ++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 78 insertions(+), 1 deletion(-) diff --git a/plugins/mod_muc_mam.lua b/plugins/mod_muc_mam.lua index ede4e57a..1fbc8b2a 100644 --- a/plugins/mod_muc_mam.lua +++ b/plugins/mod_muc_mam.lua @@ -29,7 +29,7 @@ local is_stanza = st.is_stanza; local tostring = tostring; local time_now = os.time; local m_min = math.min; -local timestamp, timestamp_parse = require "util.datetime".datetime, require "util.datetime".parse; +local timestamp, timestamp_parse, datestamp = import( "util.datetime", "datetime", "parse", "date"); local default_max_items, max_max_items = 20, module:get_option_number("max_archive_query_results", 50); local default_history_length = 20; @@ -39,6 +39,10 @@ local function get_historylength(room) return math.min(room._data.history_length or default_history_length, max_history_length); end +function schedule_cleanup() + -- replaced by non-noop later if cleanup is enabled +end + local log_all_rooms = module:get_option_boolean("muc_log_all_rooms", false); local log_by_default = module:get_option_boolean("muc_log_by_default", true); @@ -351,6 +355,7 @@ local function save_to_history(self, stanza) local id = archive:append(room_node, nil, stored_stanza, time_now(), with); if id then + schedule_cleanup(room_node); stanza:add_direct_child(st.stanza("stanza-id", { xmlns = xmlns_st_id, by = self.jid, id = id })); end end @@ -386,3 +391,75 @@ module:add_feature(xmlns_mam); module:hook("muc-disco#info", function(event) event.reply:tag("feature", {var=xmlns_mam}):up(); end); + +-- Cleanup + +local cleanup_after = module:get_option_string("muc_log_expires_after", "1w"); +local cleanup_interval = module:get_option_number("muc_log_cleanup_interval", 4 * 60 * 60); + +if cleanup_after ~= "never" then + local cleanup_storage = module:open_store("muc_log_cleanup"); + local cleanup_map = module:open_store("muc_log_cleanup", "map"); + + local day = 86400; + local multipliers = { d = day, w = day * 7, m = 31 * day, y = 365.2425 * day }; + local n, m = cleanup_after:lower():match("(%d+)%s*([dwmy]?)"); + if not n then + module:log("error", "Could not parse muc_log_expires_after string %q", cleanup_after); + return false; + end + + cleanup_after = tonumber(n) * ( multipliers[m] or 1 ); + + module:log("debug", "muc_log_expires_after = %d -- in seconds", cleanup_after); + + if not archive.delete then + module:log("error", "muc_log_expires_after set but mod_%s does not support deleting", archive._provided_by); + return false; + end + + -- For each day, store a set of rooms that have new messages. To expire + -- messages, we collect the union of sets of rooms from dates that fall + -- outside the cleanup range. + + function schedule_cleanup(roomname, date) + cleanup_map:set(date or datestamp(), roomname, true); + end + + cleanup_runner = require "util.async".runner(function () + local rooms = {}; + local cut_off = datestamp(os.time() - cleanup_after); + for date in cleanup_storage:users() do + if date <= cut_off then + module:log("debug", "Messages from %q should be expired", date); + local messages_this_day = cleanup_storage:get(date); + if messages_this_day then + for room in pairs(messages_this_day) do + rooms[room] = true; + end + if date < cut_off then + -- Messages from the same day as the cut-off might not have expired yet, + -- but all earlier will have, so clear storage for those days. + cleanup_storage:set(date, nil); + end + end + end + end + local sum, num_rooms = 0, 0; + for room in pairs(rooms) do + local ok, err = archive:delete(room, { ["end"] = os.time() - cleanup_after; }) + if ok then + num_rooms = num_rooms + 1; + sum = sum + (tonumber(ok) or 0); + end + end + module:log("info", "Deleted %d expired messages for %d rooms", sum, num_rooms); + end); + + cleanup_task = module:add_timer(1, function () + cleanup_runner:run(true); + return cleanup_interval; + end); +else + module:log("debug", "Archive expiry disabled"); +end -- cgit v1.2.3 From b6cff0ba5e41f9be9232b37461df5ff11cacfa6e Mon Sep 17 00:00:00 2001 From: Kim Alvefur Date: Mon, 14 Jan 2019 00:17:02 +0100 Subject: mod_storage_memory: Implement :user iteration API --- plugins/mod_storage_memory.lua | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/plugins/mod_storage_memory.lua b/plugins/mod_storage_memory.lua index 71205ee0..745e394b 100644 --- a/plugins/mod_storage_memory.lua +++ b/plugins/mod_storage_memory.lua @@ -23,6 +23,10 @@ local function _purge_store(self, username) return true; end +local function _users(self) + return next, self.store, nil; +end + local keyval_store = {}; keyval_store.__index = keyval_store; @@ -40,9 +44,13 @@ end keyval_store.purge = _purge_store; +keyval_store.users = _users; + local archive_store = {}; archive_store.__index = archive_store; +archive_store.users = _users; + function archive_store:append(username, key, value, when, with) if is_stanza(value) then value = st.preserialize(value); -- cgit v1.2.3