aboutsummaryrefslogtreecommitdiffstats
path: root/plugins/mod_smacks.lua
diff options
context:
space:
mode:
Diffstat (limited to 'plugins/mod_smacks.lua')
-rw-r--r--plugins/mod_smacks.lua159
1 files changed, 64 insertions, 95 deletions
diff --git a/plugins/mod_smacks.lua b/plugins/mod_smacks.lua
index 3a4c7b84..42a903f6 100644
--- a/plugins/mod_smacks.lua
+++ b/plugins/mod_smacks.lua
@@ -2,7 +2,7 @@
--
-- Copyright (C) 2010-2015 Matthew Wild
-- Copyright (C) 2010 Waqas Hussain
--- Copyright (C) 2012-2021 Kim Alvefur
+-- Copyright (C) 2012-2022 Kim Alvefur
-- Copyright (C) 2012 Thijs Alkemade
-- Copyright (C) 2014 Florian Zeitz
-- Copyright (C) 2016-2020 Thilo Molitor
@@ -10,6 +10,7 @@
-- This project is MIT/X11 licensed. Please see the
-- COPYING file in the source package for more information.
--
+-- TODO unify sendq and smqueue
local tonumber = tonumber;
local tostring = tostring;
@@ -83,6 +84,22 @@ local all_old_sessions = module:open_store("smacks_h");
local old_session_registry = module:open_store("smacks_h", "map");
local session_registry = module:shared "/*/smacks/resumption-tokens"; -- > user@host/resumption-token --> resource
+local function track_session(session, id)
+ session_registry[jid.join(session.username, session.host, id or session.resumption_token)] = session;
+ session.resumption_token = id;
+end
+
+local function save_old_session(session)
+ session_registry[jid.join(session.username, session.host, session.resumption_token)] = nil;
+ return old_session_registry:set(session.username, session.resumption_token,
+ { h = session.handled_stanza_count; t = os.time() })
+end
+
+local function clear_old_session(session, id)
+ session_registry[jid.join(session.username, session.host, id or session.resumption_token)] = nil;
+ return old_session_registry:set(session.username, id or session.resumption_token, nil)
+end
+
local ack_errors = require"util.error".init("mod_smacks", xmlns_sm3, {
head = { condition = "undefined-condition"; text = "Client acknowledged more stanzas than sent by server" };
tail = { condition = "undefined-condition"; text = "Client acknowledged less stanzas than already acknowledged" };
@@ -155,13 +172,12 @@ end
local function request_ack(session, reason)
local queue = session.outgoing_stanza_queue;
- session.log("debug", "Sending <r> (inside timer, before send) from %s - #queue=%d", reason, queue:count_unacked());
+ session.log("debug", "Sending <r> from %s - #queue=%d", reason, queue:count_unacked());
session.awaiting_ack = true;
(session.sends2s or session.send)(st.stanza("r", { xmlns = session.smacks }))
if session.destroyed then return end -- sending something can trigger destruction
-- expected_h could be lower than this expression e.g. more stanzas added to the queue meanwhile)
session.last_requested_h = queue:count_acked() + queue:count_unacked();
- session.log("debug", "Sending <r> (inside timer, after send) from %s - #queue=%d", reason, queue:count_unacked());
if not session.delayed_ack_timer then
session.delayed_ack_timer = timer.add_task(delayed_ack_timeout, function()
ack_delayed(session, nil); -- we don't know if this is the only new stanza in the queue
@@ -180,7 +196,6 @@ local function outgoing_stanza_filter(stanza, session)
-- supposed to be nil.
-- However, when using mod_smacks with mod_websocket, then mod_websocket's
-- stanzas/out filter can get called before this one and adds the xmlns.
- if session.resending_unacked then return stanza end
if not session.smacks then return stanza end
local is_stanza = st.is_stanza(stanza) and
(not stanza.attr.xmlns or stanza.attr.xmlns == 'jabber:client')
@@ -234,8 +249,7 @@ module:hook("pre-session-close", function(event)
if session.smacks == nil then return end
if session.resumption_token then
session.log("debug", "Revoking resumption token");
- session_registry[jid.join(session.username, session.host, session.resumption_token)] = nil;
- old_session_registry:set(session.username, session.resumption_token, nil);
+ clear_old_session(session);
session.resumption_token = nil;
else
session.log("debug", "Session not resumable");
@@ -284,7 +298,7 @@ function handle_enable(session, stanza, xmlns_sm)
if session.username then
local old_sessions, err = all_old_sessions:get(session.username);
- module:log("debug", "Old sessions: %q", old_sessions)
+ session.log("debug", "Old sessions: %q", old_sessions)
if old_sessions then
local keep, count = {}, 0;
for token, info in it.sorted_pairs(old_sessions, function(a, b)
@@ -296,11 +310,11 @@ function handle_enable(session, stanza, xmlns_sm)
end
all_old_sessions:set(session.username, keep);
elseif err then
- module:log("error", "Unable to retrieve old resumption counters: %s", err);
+ session.log("error", "Unable to retrieve old resumption counters: %s", err);
end
end
- module:log("debug", "Enabling stream management");
+ session.log("debug", "Enabling stream management");
session.smacks = xmlns_sm;
wrap_session(session, false);
@@ -310,8 +324,7 @@ function handle_enable(session, stanza, xmlns_sm)
local resume = stanza.attr.resume;
if resume == "true" or resume == "1" then
resume_token = new_id();
- session_registry[jid.join(session.username, session.host, resume_token)] = session;
- session.resumption_token = resume_token;
+ track_session(session, resume_token);
resume_max = tostring(resume_timeout);
end
(session.sends2s or session.send)(st.stanza("enabled", { xmlns = xmlns_sm, id = resume_token, resume = resume, max = resume_max }));
@@ -320,29 +333,23 @@ end
module:hook_tag(xmlns_sm2, "enable", function (session, stanza) return handle_enable(session, stanza, xmlns_sm2); end, 100);
module:hook_tag(xmlns_sm3, "enable", function (session, stanza) return handle_enable(session, stanza, xmlns_sm3); end, 100);
-module:hook_tag("http://etherx.jabber.org/streams", "features",
- function (session, stanza)
- -- Needs to be done after flushing sendq since those aren't stored as
- -- stanzas and counting them is weird.
- -- TODO unify sendq and smqueue
- timer.add_task(1e-6, function ()
- if can_do_smacks(session) then
- if stanza:get_child("sm", xmlns_sm3) then
- session.sends2s(st.stanza("enable", sm3_attr));
- session.smacks = xmlns_sm3;
- elseif stanza:get_child("sm", xmlns_sm2) then
- session.sends2s(st.stanza("enable", sm2_attr));
- session.smacks = xmlns_sm2;
- else
- return;
- end
- wrap_session_out(session, false);
- end
- end);
- end);
+module:hook_tag("http://etherx.jabber.org/streams", "features", function(session, stanza)
+ if can_do_smacks(session) then
+ session.smacks_feature = stanza:get_child("sm", xmlns_sm3) or stanza:get_child("sm", xmlns_sm2);
+ end
+end);
+
+module:hook("s2sout-established", function (event)
+ local session = event.session;
+ if not session.smacks_feature then return end
+
+ session.smacks = session.smacks_feature.attr.xmlns;
+ wrap_session_out(session, false);
+ session.sends2s(st.stanza("enable", { xmlns = session.smacks }));
+end);
function handle_enabled(session, stanza, xmlns_sm) -- luacheck: ignore 212/stanza
- module:log("debug", "Enabling stream management");
+ session.log("debug", "Enabling stream management");
session.smacks = xmlns_sm;
wrap_session_in(session, false);
@@ -356,10 +363,10 @@ module:hook_tag(xmlns_sm3, "enabled", function (session, stanza) return handle_e
function handle_r(origin, stanza, xmlns_sm) -- luacheck: ignore 212/stanza
if not origin.smacks then
- module:log("debug", "Received ack request from non-smack-enabled session");
+ origin.log("debug", "Received ack request from non-smack-enabled session");
return;
end
- module:log("debug", "Received ack request, acking for %d", origin.handled_stanza_count);
+ origin.log("debug", "Received ack request, acking for %d", origin.handled_stanza_count);
-- Reply with <a>
(origin.sends2s or origin.send)(st.stanza("a", { xmlns = xmlns_sm, h = format_h(origin.handled_stanza_count) }));
-- piggyback our own ack request if needed (see request_ack_if_needed() for explanation of last_requested_h)
@@ -412,13 +419,14 @@ local function handle_unacked_stanzas(session)
local queue = session.outgoing_stanza_queue;
local unacked = queue:count_unacked()
if unacked > 0 then
+ local error_from = jid.join(session.username, session.host or module.host);
tx_dropped_stanzas:sample(unacked);
session.smacks = false; -- Disable queueing
session.outgoing_stanza_queue = nil;
for stanza in queue._queue:consume() do
if not module:fire_event("delivery/failure", { session = session, stanza = stanza }) then
if stanza.attr.type ~= "error" and stanza.attr.from ~= session.full_jid then
- local reply = st.error_reply(stanza, "cancel", "recipient-unavailable");
+ local reply = st.error_reply(stanza, "cancel", "recipient-unavailable", nil, error_from);
module:send(reply);
end
end
@@ -485,11 +493,8 @@ module:hook("pre-resource-unbind", function (event)
end
session.log("debug", "Destroying session for hibernating too long");
- session_registry[jid.join(session.username, session.host, session.resumption_token)] = nil;
- old_session_registry:set(session.username, session.resumption_token,
- { h = session.handled_stanza_count; t = os.time() });
+ save_old_session(session);
session.resumption_token = nil;
- session.resending_unacked = true; -- stop outgoing_stanza_filter from re-queueing anything anymore
sessionmanager.destroy_session(session, "Hibernating too long");
sessions_expired(1);
end);
@@ -522,10 +527,6 @@ end
module:hook("s2sout-destroyed", handle_s2s_destroyed);
module:hook("s2sin-destroyed", handle_s2s_destroyed);
-local function get_session_id(session)
- return session.id or (tostring(session):match("[a-f0-9]+$"));
-end
-
function handle_resume(session, stanza, xmlns_sm)
if session.full_jid then
session.log("warn", "Tried to resume after resource binding");
@@ -544,7 +545,7 @@ function handle_resume(session, stanza, xmlns_sm)
session.send(st.stanza("failed", { xmlns = xmlns_sm, h = format_h(old_session.h) })
:tag("item-not-found", { xmlns = xmlns_errors })
);
- old_session_registry:set(session.username, id, nil);
+ clear_old_session(session, id);
resumption_expired(1);
else
session.log("debug", "Tried to resume non-existent session with id %s", id);
@@ -566,40 +567,8 @@ function handle_resume(session, stanza, xmlns_sm)
local now = os_time();
age = now - original_session.hibernating;
end
- session.log("debug", "mod_smacks resuming existing session %s...", get_session_id(original_session));
- original_session.log("debug", "mod_smacks session resumed from %s...", get_session_id(session));
- -- TODO: All this should move to sessionmanager (e.g. session:replace(new_session))
- if original_session.conn then
- original_session.log("debug", "mod_smacks closing an old connection for this session");
- local conn = original_session.conn;
- c2s_sessions[conn] = nil;
- conn:close();
- end
- local migrated_session_log = session.log;
- original_session.ip = session.ip;
- original_session.conn = session.conn;
- original_session.rawsend = session.rawsend;
- original_session.rawsend.session = original_session;
- original_session.rawsend.conn = original_session.conn;
- original_session.send = session.send;
- original_session.send.session = original_session;
- original_session.close = session.close;
- original_session.filter = session.filter;
- original_session.filter.session = original_session;
- original_session.filters = session.filters;
- original_session.send.filter = original_session.filter;
- original_session.stream = session.stream;
- original_session.secure = session.secure;
- original_session.hibernating = nil;
- original_session.resumption_counter = (original_session.resumption_counter or 0) + 1;
- session.log = original_session.log;
- session.type = original_session.type;
- wrap_session(original_session, true);
- -- Inform xmppstream of the new session (passed to its callbacks)
- original_session.stream:set_session(original_session);
- -- Similar for connlisteners
- c2s_sessions[session.conn] = original_session;
+ session.log("debug", "mod_smacks resuming existing session %s...", original_session.id);
local queue = original_session.outgoing_stanza_queue;
local h = tonumber(stanza.attr.h);
@@ -611,13 +580,17 @@ function handle_resume(session, stanza, xmlns_sm)
err = ack_errors.new("overflow");
end
- if err or not queue:resumable() then
- original_session.send(st.stanza("failed",
+ if err then
+ session.send(st.stanza("failed",
{ xmlns = xmlns_sm; h = format_h(original_session.handled_stanza_count); previd = id }));
- original_session:close(err);
- return false;
+ session.log("debug", "Resumption failed: %s", err);
+ return true;
end
+ -- Update original_session with the parameters (connection, etc.) from the new session
+ sessionmanager.update_session(original_session, session);
+
+ -- Inform client of successful resumption
original_session.send(st.stanza("resumed", { xmlns = xmlns_sm,
h = format_h(original_session.handled_stanza_count), previd = id }));
@@ -626,20 +599,16 @@ function handle_resume(session, stanza, xmlns_sm)
-- We have to use the send of "session" because we don't want to add our resent stanzas
-- to the outgoing queue again
- session.log("debug", "resending all unacked stanzas that are still queued after resume, #queue = %d", queue:count_unacked());
- -- FIXME Which session is it that the queue filter sees?
- session.resending_unacked = true;
- original_session.resending_unacked = true;
+ original_session.log("debug", "resending all unacked stanzas that are still queued after resume, #queue = %d", queue:count_unacked());
for _, queued_stanza in queue:resume() do
- session.send(queued_stanza);
- end
- session.resending_unacked = nil;
- original_session.resending_unacked = nil;
- session.log("debug", "all stanzas resent, now disabling send() in this migrated session, #queue = %d", queue:count_unacked());
- function session.send(stanza) -- luacheck: ignore 432
- migrated_session_log("error", "Tried to send stanza on old session migrated by smacks resume (maybe there is a bug?): %s", tostring(stanza));
- return false;
+ original_session.send(queued_stanza);
end
+ session.log("debug", "all stanzas resent, enabling stream management on resumed stream, #queue = %d", queue:count_unacked());
+
+ -- Add our own handlers to the resumed session (filters have been reset in the update)
+ wrap_session(original_session, true);
+
+ -- Let everyone know that we are no longer hibernating
module:fire_event("smacks-hibernation-end", {origin = session, resumed = original_session, queue = queue:table()});
original_session.awaiting_ack = nil; -- Don't wait for acks from before the resumption
request_ack_now_if_needed(original_session, true, "handle_resume", nil);
@@ -647,6 +616,7 @@ function handle_resume(session, stanza, xmlns_sm)
end
return true;
end
+
module:hook_tag(xmlns_sm2, "resume", function (session, stanza) return handle_resume(session, stanza, xmlns_sm2); end);
module:hook_tag(xmlns_sm3, "resume", function (session, stanza) return handle_resume(session, stanza, xmlns_sm3); end);
@@ -701,8 +671,7 @@ module:hook_global("server-stopping", function(event)
for _, user in pairs(local_sessions) do
for _, session in pairs(user.sessions) do
if session.resumption_token then
- if old_session_registry:set(session.username, session.resumption_token,
- { h = session.handled_stanza_count; t = os.time() }) then
+ if save_old_session(session) then
session.resumption_token = nil;
-- Deal with unacked stanzas
generated by cgit v1.2.3 (git 2.25.1) at 2024-10-01 23:55:56 +0000