# HG changeset patch # User tmolitor # Date 1488741833 -3600 # Node ID ffb6646b4253e49f56fa0115a153f2d80660435c # Parent 307ddebb72e123cc97b463e0f92e9ec51ac5909c Implement XEP-0198 revision 1.5.2 and limit number of hibernated sessions per user Revision 1.5.2 allows sending h-values on resumes that fail due to hibernation timeout and to send out a smacks ack directly before the stream close tag. I also made the used timers stoppable even for prosody 0.10 and below, this makes the smacks-ack-delayed event more useful. diff -r 307ddebb72e1 -r ffb6646b4253 mod_smacks/README.markdown --- a/mod_smacks/README.markdown Sat Mar 04 19:52:41 2017 +0100 +++ b/mod_smacks/README.markdown Sun Mar 05 20:23:53 2017 +0100 @@ -39,24 +39,31 @@ "smacks-hibernation-end". See [mod_cloud_notify] for details on how this events are used there. +Use prosody 0.10+ to have per user limits on allowed sessions in hibernation +state and allowed sessions for which the h-value is kept even after the +hibernation timed out. +These are settable using `smacks_max_hibernated_sessions` and `smacks_max_old_sessions`. + Configuration ============= - Option Default Description - ------------------------------ ----------------- ----------------------------------------------------------------------------------------- - `smacks_hibernation_time` 300 (5 minutes) The number of seconds a disconnected session should stay alive for (to allow reconnect) - `smacks_enabled_s2s` false Enable Stream Management on server connections? *Experimental* - `smacks_max_unacked_stanzas` 0 How many stanzas to send before requesting acknowledgement - `smacks_max_ack_delay` 60 (1 minute) The number of seconds an ack must be unanswered to trigger an "smacks-ack-delayed" event + Option Default Description + ---------------------------------- ----------------- ------------------------------------------------------------------------------------------------------------------ + `smacks_hibernation_time` 300 (5 minutes) The number of seconds a disconnected session should stay alive for (to allow reconnect) + `smacks_enabled_s2s` false Enable Stream Management on server connections? *Experimental* + `smacks_max_unacked_stanzas` 0 How many stanzas to send before requesting acknowledgement + `smacks_max_ack_delay` 60 (1 minute) The number of seconds an ack must be unanswered to trigger an "smacks-ack-delayed" event + `smacks_max_hibernated_sessions` 10 The number of allowed sessions in hibernated state (limited per user) + `smacks_max_old_sessions` 10 The number of allowed sessions with timed out hibernation for which the h-value is still kept (limited per user) Compatibility ============= - ----- ----------------------------------- + ----- ----------------------------------------------------------------------------- 0.10 Works - 0.9 Works - 0.8 Works, use version [7693724881b3] - ----- ----------------------------------- + 0.9 Works, no per user limit of hibernated sessions + 0.8 Works, no per user limit of hibernated sessions, use version [7693724881b3] + ----- ----------------------------------------------------------------------------- Clients @@ -64,11 +71,13 @@ Clients that support [XEP-0198]: -- Gajim +- Gajim (Linux, Windows, OS X) +- Conversations (Android) +- ChatSecure (iOS) - Swift (but not resumption, as of version 2.0 and alphas of 3.0) - Psi (in an unreleased branch) -- Conversations -- Yaxim +- Yaxim (Android) +- Monal (iOS) [7693724881b3]: //hg.prosody.im/prosody-modules/raw-file/7693724881b3/mod_smacks/mod_smacks.lua [mod_smacks_offline]: //modules.prosody.im/mod_smacks_offline diff -r 307ddebb72e1 -r ffb6646b4253 mod_smacks/mod_smacks.lua --- a/mod_smacks/mod_smacks.lua Sat Mar 04 19:52:41 2017 +0100 +++ b/mod_smacks/mod_smacks.lua Sun Mar 05 20:23:53 2017 +0100 @@ -12,6 +12,8 @@ -- local st = require "util.stanza"; +local dep = require "util.dependencies"; +local cache = dep.softreq("util.cache"); -- only available in prosody 0.10+ local uuid_generate = require "util.uuid".generate; local t_insert, t_remove = table.insert, table.remove; @@ -35,16 +37,71 @@ local s2s_resend = module:get_option_boolean("smacks_s2s_resend", false); local max_unacked_stanzas = module:get_option_number("smacks_max_unacked_stanzas", 0); local delayed_ack_timeout = module:get_option_number("smacks_max_ack_delay", 60); +local max_hibernated_sessions = module:get_option_number("smacks_max_hibernated_sessions", 10); +local max_old_sessions = module:get_option_number("smacks_max_old_sessions", 10); local core_process_stanza = prosody.core_process_stanza; local sessionmanager = require"core.sessionmanager"; local c2s_sessions = module:shared("/*/c2s/sessions"); -local session_registry = {}; + +local function init_session_cache(max_entries, evict_callback) + -- old prosody version < 0.10 (no limiting at all!) + if not cache then + local store = {}; + return { + get = function(user, key) return store[user.."@"..key]; end; + set = function(user, key, value) store[user.."@"..key] = value; end; + }; + end + + -- use per user limited cache for prosody >= 0.10 + local stores = {}; + return { + get = function(user, key) + if not stores[user] then + stores[user] = cache.new(max_entries, evict_callback); + end + return stores[user]:get(key); + end; + set = function(user, key, value) + if not stores[user] then stores[user] = cache.new(max_entries, evict_callback); end + stores[user]:set(key, value); + -- remove empty caches completely + if not stores[user]:count() then stores[user] = nil; end + end; + }; +end +local old_session_registry = init_session_cache(max_old_sessions, nil); +local session_registry = init_session_cache(max_hibernated_sessions, function(resumption_token, session) + if session.destroyed then return; end + session.log("warn", "User has too much hibernated sessions, removing oldest session (token: %s)", resumption_token); + -- store old session's h values on force delete + -- save only actual h value and username/host (for security) + old_session_registry.set(session.username, resumption_token, { + h = session.handled_stanza_count, + username = session.username, + host = session.host + }); + return true; -- allow session to be removed from full cache to make room for new one +end); + +local function stoppable_timer(delay, callback) + local stopped = false; + return { + stop = function () stopped = true end; + module:add_timer(delay, function (t) + if stopped then return; end + return callback(t); + end); + }; +end local function delayed_ack_function(session) - -- fire event only when configured to do so - if delayed_ack_timeout > 0 and session.awaiting_ack and not (session.outgoing_stanza_queue == nil) then - session.log("debug", "Firing event 'smacks-ack-delayed', queue = %d", #session.outgoing_stanza_queue); + -- fire event only if configured to do so and our session is not hibernated or destroyed + if delayed_ack_timeout > 0 and session.awaiting_ack + and not session.hibernating and not session.destroyed then + session.log("debug", "Firing event 'smacks-ack-delayed', queue = %d", + session.outgoing_stanza_queue and #session.outgoing_stanza_queue or 0); module:fire_event("smacks-ack-delayed", {origin = session, queue = session.outgoing_stanza_queue}); end session.delayed_ack_timer = nil; @@ -86,15 +143,17 @@ if #queue > max_unacked_stanzas and session.awaiting_ack == nil then session.log("debug", "Queuing (in a moment)"); session.awaiting_ack = false; - session.awaiting_ack_timer = module:add_timer(1e-06, function () + session.awaiting_ack_timer = stoppable_timer(1e-06, function () if not session.awaiting_ack then session.log("debug", "Sending (inside timer, before send)"); (session.sends2s or session.send)(st.stanza("r", { xmlns = session.smacks })) session.log("debug", "Sending (inside timer, after send)"); session.awaiting_ack = true; - session.delayed_ack_timer = module:add_timer(delayed_ack_timeout, function() - delayed_ack_function(session); - end); + if not session.delayed_ack_timer then + session.delayed_ack_timer = stoppable_timer(delayed_ack_timeout, function() + delayed_ack_function(session); + end); + end end end); end @@ -149,9 +208,14 @@ local session_close = session.close; function session.close(...) if session.resumption_token then - session_registry[session.resumption_token] = nil; + session_registry.set(session.username, session.resumption_token, nil); + old_session_registry.set(session.username, session.resumption_token, nil); session.resumption_token = nil; end + -- send out last ack as per revision 1.5.2 of XEP-0198 + if session.smacks then + (session.sends2s or session.send)(st.stanza("a", { xmlns = session.smacks, h = tostring(session.handled_stanza_count) })); + end return session_close(...); end return session; @@ -189,7 +253,7 @@ local resume = stanza.attr.resume; if resume == "true" or resume == "1" then resume_token = uuid_generate(); - session_registry[resume_token] = session; + session_registry.set(session.username, resume_token, session); session.resumption_token = resume_token; end (session.sends2s or session.send)(st.stanza("enabled", { xmlns = xmlns_sm, id = resume_token, resume = resume })); @@ -200,7 +264,7 @@ module:hook_stanza("http://etherx.jabber.org/streams", "features", function (session, stanza) - module:add_timer(1e-6, function () + stoppable_timer(1e-6, function () if can_do_smacks(session) then if stanza:get_child("sm", xmlns_sm3) then session.sends2s(st.stanza("enable", sm3_attr)); @@ -253,7 +317,7 @@ origin.delayed_ack_timer = nil; end -- Remove handled stanzas from outgoing_stanza_queue - log("debug", "ACK: h=%s, last=%s", stanza.attr.h or "", origin.last_acknowledged_stanza or ""); + -- origin.log("debug", "ACK: h=%s, last=%s", stanza.attr.h or "", origin.last_acknowledged_stanza or ""); local h = tonumber(stanza.attr.h); if not h then origin:close{ condition = "invalid-xml"; text = "Missing or invalid 'h' attribute"; }; @@ -330,7 +394,13 @@ -- otherwise the session resumed and re-hibernated. and session.hibernating == hibernate_time then session.log("debug", "Destroying session for hibernating too long"); - session_registry[session.resumption_token] = nil; + session_registry.set(session.username, session.resumption_token, nil); + -- save only actual h value and username/host (for security) + old_session_registry.set(session.username, session.resumption_token, { + h = session.handled_stanza_count, + username = session.username, + host = session.host + }); session.resumption_token = nil; sessionmanager.destroy_session(session); else @@ -372,12 +442,21 @@ end local id = stanza.attr.previd; - local original_session = session_registry[id]; + local original_session = session_registry.get(session.username, id); if not original_session then session.log("debug", "Tried to resume non-existent session with id %s", id); - session.send(st.stanza("failed", { xmlns = xmlns_sm }) - :tag("item-not-found", { xmlns = xmlns_errors }) - ); + local old_session = old_session_registry.get(session.username, id); + if old_session and session.username == old_session.username + and session.host == old_session.host + and old_session.h then + session.send(st.stanza("failed", { xmlns = xmlns_sm, h = tostring(old_session.h) }) + :tag("item-not-found", { xmlns = xmlns_errors }) + ); + else + session.send(st.stanza("failed", { xmlns = xmlns_sm }) + :tag("item-not-found", { xmlns = xmlns_errors }) + ); + end; elseif session.username == original_session.username and session.host == original_session.host then session.log("debug", "mod_smacks resuming existing session..."); @@ -448,9 +527,11 @@ session.awaiting_ack = false; (session.sends2s or session.send)(st.stanza("r", { xmlns = session.smacks })); session.awaiting_ack = true; - session.delayed_ack_timer = module:add_timer(delayed_ack_timeout, function() - delayed_ack_function(session); - end); + if not session.delayed_ack_timer then + session.delayed_ack_timer = stoppable_timer(delayed_ack_timeout, function() + delayed_ack_function(session); + end); + end return true; end end