prosody
464a22f2751c
plugins/mod_smacks.lua

Find changesets by keywords by author, files, the commit message, revision number or hash, or revset expression.

Software / code / prosody

Comparison

plugins/mod_smacks.lua @ 12682:464a22f2751c

mod_smacks: Split resumption into multiple stages, to simplify ISR integration This will allow us to return the success/failed as part of the SASL2 response, and *then* perform the stanza sync as a second step.
author Matthew Wild <mwild1@gmail.com>
date Fri, 26 Aug 2022 19:07:36 +0100
parent 12678:5a61e1603f42
child 12688:36ba170c4fd0
comparison
equal deleted inserted replaced
12681:16ea01745dbe 12682:464a22f2751c
105 tail = { condition = "undefined-condition"; text = "Client acknowledged less stanzas than already acknowledged" }; 105 tail = { condition = "undefined-condition"; text = "Client acknowledged less stanzas than already acknowledged" };
106 pop = { condition = "internal-server-error"; text = "Something went wrong with Stream Management" }; 106 pop = { condition = "internal-server-error"; text = "Something went wrong with Stream Management" };
107 overflow = { condition = "resource-constraint", text = "Too many unacked stanzas remaining, session can't be resumed" } 107 overflow = { condition = "resource-constraint", text = "Too many unacked stanzas remaining, session can't be resumed" }
108 }); 108 });
109 109
110 local resume_errors = require "util.error".init("mod_smacks", xmlns_sm3, {
111 expired = { condition = "item-not-found", text = "Session expired, and cannot be resumed" };
112 already_bound = { condition = "unexpected-request", text = "Cannot resume another session after a resource is bound" };
113 unknown_session = { condition = "item-not-found", text = "Unknown session" };
114 });
115
110 -- COMPAT note the use of compatibility wrapper in events (queue:table()) 116 -- COMPAT note the use of compatibility wrapper in events (queue:table())
111 117
112 local function ack_delayed(session, stanza) 118 local function ack_delayed(session, stanza)
113 -- fire event only if configured to do so and our session is not already hibernated or destroyed 119 -- fire event only if configured to do so and our session is not already hibernated or destroyed
114 if delayed_ack_timeout > 0 and session.awaiting_ack 120 if delayed_ack_timeout > 0 and session.awaiting_ack
525 end 531 end
526 532
527 module:hook("s2sout-destroyed", handle_s2s_destroyed); 533 module:hook("s2sout-destroyed", handle_s2s_destroyed);
528 module:hook("s2sin-destroyed", handle_s2s_destroyed); 534 module:hook("s2sin-destroyed", handle_s2s_destroyed);
529 535
530 function handle_resume(session, stanza, xmlns_sm) 536 function do_resume(session, stanza)
531 if session.full_jid then 537 if session.full_jid then
532 session.log("warn", "Tried to resume after resource binding"); 538 session.log("warn", "Tried to resume after resource binding");
533 session.send(st.stanza("failed", { xmlns = xmlns_sm }) 539 return nil, resume_errors.new("already_bound");
534 :tag("unexpected-request", { xmlns = xmlns_errors })
535 );
536 return true;
537 end 540 end
538 541
539 local id = stanza.attr.previd; 542 local id = stanza.attr.previd;
540 local original_session = session_registry[jid.join(session.username, session.host, id)]; 543 local original_session = session_registry[jid.join(session.username, session.host, id)];
541 if not original_session then 544 if not original_session then
542 local old_session = old_session_registry:get(session.username, id); 545 local old_session = old_session_registry:get(session.username, id);
543 if old_session then 546 if old_session then
544 session.log("debug", "Tried to resume old expired session with id %s", id); 547 session.log("debug", "Tried to resume old expired session with id %s", id);
545 session.send(st.stanza("failed", { xmlns = xmlns_sm, h = format_h(old_session.h) })
546 :tag("item-not-found", { xmlns = xmlns_errors })
547 );
548 clear_old_session(session, id); 548 clear_old_session(session, id);
549 resumption_expired(1); 549 resumption_expired(1);
550 else 550 return nil, resume_errors.new("expired", { h = old_session.h });
551 session.log("debug", "Tried to resume non-existent session with id %s", id); 551 end
552 session.send(st.stanza("failed", { xmlns = xmlns_sm }) 552 session.log("debug", "Tried to resume non-existent session with id %s", id);
553 :tag("item-not-found", { xmlns = xmlns_errors }) 553 return nil, resume_errors.new("unknown_session");
554 ); 554 end
555
556 if original_session.hibernating_watchdog then
557 original_session.log("debug", "Letting the watchdog go");
558 original_session.hibernating_watchdog:cancel();
559 original_session.hibernating_watchdog = nil;
560 elseif session.hibernating then
561 original_session.log("error", "Hibernating session has no watchdog!")
562 end
563 -- zero age = was not hibernating yet
564 local age = 0;
565 if original_session.hibernating then
566 local now = os_time();
567 age = now - original_session.hibernating;
568 end
569
570 session.log("debug", "mod_smacks resuming existing session %s...", original_session.id);
571
572 local queue = original_session.outgoing_stanza_queue;
573 local h = tonumber(stanza.attr.h);
574
575 original_session.log("debug", "Pre-resumption #queue = %d", queue:count_unacked())
576 local acked, err = ack_errors.coerce(queue:ack(h)); -- luacheck: ignore 211/acked
577
578 if not err and not queue:resumable() then
579 err = ack_errors.new("overflow");
580 end
581
582 if err then
583 session.log("debug", "Resumption failed: %s", err);
584 return nil, err;
585 end
586
587 -- Update original_session with the parameters (connection, etc.) from the new session
588 sessionmanager.update_session(original_session, session);
589
590 return {
591 session = original_session;
592 id = id;
593 -- Return function to complete the resumption and resync unacked stanzas
594 -- This is two steps so we can support SASL2/ISR
595 finish = function ()
596 -- Ok, we need to re-send any stanzas that the client didn't see
597 -- ...they are what is now left in the outgoing stanza queue
598 -- We have to use the send of "session" because we don't want to add our resent stanzas
599 -- to the outgoing queue again
600
601 original_session.log("debug", "resending all unacked stanzas that are still queued after resume, #queue = %d", queue:count_unacked());
602 for _, queued_stanza in queue:resume() do
603 original_session.send(queued_stanza);
604 end
605 original_session.log("debug", "all stanzas resent, enabling stream management on resumed stream, #queue = %d", queue:count_unacked());
606
607 -- Add our own handlers to the resumed session (filters have been reset in the update)
608 wrap_session(original_session, true);
609
610 -- Let everyone know that we are no longer hibernating
611 module:fire_event("smacks-hibernation-end", {origin = session, resumed = original_session, queue = queue:table()});
612 original_session.awaiting_ack = nil; -- Don't wait for acks from before the resumption
613 request_ack_now_if_needed(original_session, true, "handle_resume", nil);
614 resumption_age:sample(age);
555 end; 615 end;
556 else 616 };
557 if original_session.hibernating_watchdog then 617 end
558 original_session.log("debug", "Letting the watchdog go"); 618
559 original_session.hibernating_watchdog:cancel(); 619 function handle_resume(session, stanza, xmlns_sm)
560 original_session.hibernating_watchdog = nil; 620 local resumed, err = do_resume(session, stanza);
561 elseif session.hibernating then 621 if not resumed then
562 original_session.log("error", "Hibernating session has no watchdog!") 622 session.send(st.stanza("failed", { xmlns = xmlns_sm, h = format_h(err.context.h) })
563 end 623 :tag(err.condition, { xmlns = xmlns_errors }));
564 -- zero age = was not hibernating yet 624 return true;
565 local age = 0; 625 end
566 if original_session.hibernating then 626
567 local now = os_time(); 627 session = resumed.session;
568 age = now - original_session.hibernating; 628
569 end 629 -- Inform client of successful resumption
570 630 session.send(st.stanza("resumed", { xmlns = xmlns_sm,
571 session.log("debug", "mod_smacks resuming existing session %s...", original_session.id); 631 h = format_h(session.handled_stanza_count), previd = resumed.id }));
572 632
573 local queue = original_session.outgoing_stanza_queue; 633 -- Complete resume (sync stanzas, etc.)
574 local h = tonumber(stanza.attr.h); 634 resumed.finish();
575 635
576 original_session.log("debug", "Pre-resumption #queue = %d", queue:count_unacked())
577 local acked, err = ack_errors.coerce(queue:ack(h)); -- luacheck: ignore 211/acked
578
579 if not err and not queue:resumable() then
580 err = ack_errors.new("overflow");
581 end
582
583 if err then
584 session.send(st.stanza("failed",
585 { xmlns = xmlns_sm; h = format_h(original_session.handled_stanza_count); previd = id }));
586 session.log("debug", "Resumption failed: %s", err);
587 return true;
588 end
589
590 -- Update original_session with the parameters (connection, etc.) from the new session
591 sessionmanager.update_session(original_session, session);
592
593 -- Inform client of successful resumption
594 original_session.send(st.stanza("resumed", { xmlns = xmlns_sm,
595 h = format_h(original_session.handled_stanza_count), previd = id }));
596
597 -- Ok, we need to re-send any stanzas that the client didn't see
598 -- ...they are what is now left in the outgoing stanza queue
599 -- We have to use the send of "session" because we don't want to add our resent stanzas
600 -- to the outgoing queue again
601
602 original_session.log("debug", "resending all unacked stanzas that are still queued after resume, #queue = %d", queue:count_unacked());
603 for _, queued_stanza in queue:resume() do
604 original_session.send(queued_stanza);
605 end
606 session.log("debug", "all stanzas resent, enabling stream management on resumed stream, #queue = %d", queue:count_unacked());
607
608 -- Add our own handlers to the resumed session (filters have been reset in the update)
609 wrap_session(original_session, true);
610
611 -- Let everyone know that we are no longer hibernating
612 module:fire_event("smacks-hibernation-end", {origin = session, resumed = original_session, queue = queue:table()});
613 original_session.awaiting_ack = nil; -- Don't wait for acks from before the resumption
614 request_ack_now_if_needed(original_session, true, "handle_resume", nil);
615 resumption_age:sample(age);
616 end
617 return true; 636 return true;
618 end 637 end
619 638
620 module:hook_tag(xmlns_sm2, "resume", function (session, stanza) return handle_resume(session, stanza, xmlns_sm2); end); 639 module:hook_tag(xmlns_sm2, "resume", function (session, stanza) return handle_resume(session, stanza, xmlns_sm2); end);
621 module:hook_tag(xmlns_sm3, "resume", function (session, stanza) return handle_resume(session, stanza, xmlns_sm3); end); 640 module:hook_tag(xmlns_sm3, "resume", function (session, stanza) return handle_resume(session, stanza, xmlns_sm3); end);