ServerCore/SubscriptionManager.cpp

#include "SubscriptionManager.h"
#include "Log.h"
#include "Subscription.h"
#include "SubscriptionHandlerFactory.h"
#include "TCPServer.h"
#include <algorithm>
#include <mutex>

namespace core
{

   int SubscriptionManager::add(Subscription &subscription)
   {
      lock.lock();
      subscriptions.insert(std::make_pair(subscription.id, &subscription));
      lock.unlock();
      return 1;
   }

   bool SubscriptionManager::onClearSubscription(coreutils::MString temp, coreutils::MString key)
   {
      temp = "";
      key = "";
      return true;
   }

   int SubscriptionManager::removeSessionSubscriptions(TCPSession &session)
   {
      int countSubscribed = 0;
      int countPublished = 0;
      int count = 0;

      lock.lock();
      coreutils::MString temp = "";
      for (auto [key, subscription] : subscriptions)
      {
         if (temp != "")
         {
            subscriptions.erase(temp);
            temp = "";
         }
         countSubscribed += subscription->unsubscribe(session);
         if (subscription->subscriptionOwners == coreutils::MString(session.alias))
         {
            if (subscription->selectionOfOwnerForSubscription == "*DIE")
            {
               onClearSubscription(temp, key);
            }

            else if (subscription->will == "*NEXT")
            {
               if (subscription->subscribers.size() < 1)
               {
                  coreutils::Log(coreutils::LOG_DEBUG_1) << "There is not enough people to move to the next";
                  onClearSubscription(temp, key);
               }
               else
               {
                  for (auto subscribed = subscription->subscribers.begin(); subscribed < subscription->subscribers.end(); subscribed++)
                  {
                     // pin() before reading .alias - a raw subscriber
                     // pointer here could already be mid-destruction or
                     // fully freed on another thread (see
                     // TCPSession::pin()'s own doc comment).
                     if (auto safe = (*subscribed)->pin())
                        if (safe->alias == session.alias)
                        {
                           subscription->subscribers.erase(subscribed++);
                        }
                     subscription->subscriptionOwner(session.alias, "*NEXT");
                  }
               }
            }
            else
            {
               coreutils::Log(coreutils::LOG_DEBUG_1) << "Subscription doesn't exist";
            }
            ++countPublished;
         }
      }
      if (temp != "")
      {
         subscriptions.erase(temp);
         temp = "";
      }
      coreutils::Log(coreutils::LOG_DEBUG_2) << "Removed session from " << countSubscribed << " subscription(s).";
      coreutils::Log(coreutils::LOG_DEBUG_2) << "Cancelled " << countPublished << " channel(s) for session.";
      lock.unlock();
      return countSubscribed;
   }

   int SubscriptionManager::processCommand(coreutils::ZString &request, TCPSession &session)
   {
      // Every branch below reads or mutates either the `subscriptions` map
      // or some Subscription's own `subscribers` vector - both of which
      // removeSessionSubscriptions() (called from ~TCPSession(), on
      // whatever thread happens to be tearing down a *different*
      // connection) also reads/mutates, under `lock`. This method used to
      // take no lock at all, so a session disconnecting on one thread
      // could run subscribers.erase() concurrently with subscribe()/
      // event() iterating or push_back()-ing the same vector on another -
      // an unsynchronized data race. Confirmed live: rapid connect/
      // subscribe/disconnect cycles against ItemDataBase's "createItem"
      // subscription produced a real segfault (gdb: Socket::write() ->
      // resetSocket() dereferencing a corrupted EPoll reference) while
      // broadcasting to *SUBSCRIBERS. `lock` is a plain, non-recursive
      // std::mutex, and nothing reachable from this function re-enters
      // processCommand()/add()/removeSessionSubscriptions() on the same
      // thread, so holding it for the whole call is safe.
      std::lock_guard<std::mutex> guard(lock);

      if (request[0] == "publish")
      {
         SubscriptionHandler *handler = NULL;
         if (request.getList().size() > 3)
         {
            handler = factory->getSubscriptionHandler(request[4].str());
         }
         if (request[2] == "*OWNER")
         {
            newSubscription = new Subscription(request[1].str(), session.alias, request[2].str(), request[3].str(), handler, request[5].str(), request[6].str());
            subscriptions.insert(std::make_pair(request[1].str(), newSubscription));
            return 1;
         }

         return 1;
      }
      else if (request[0] == "catalog")
      {
         session.out << ":catalog:";
         for (auto const &[key, subscription] : subscriptions)
         {
            session.out << subscription->id << ";";
         }
         session.out << std::endl;
         return 1;
      }
      else if (request[0] == "invite")
      {
         std::stringstream out;
         coreutils::Log(coreutils::LOG_DEBUG_1) << request[2];
         std::string invitee = request[2].str();
         // Was getSessionByAlias(&invitee) - getSessionByAlias(void*)
         // compares by POINTER IDENTITY against each session's own
         // alias, but &invitee is the address of a local variable that
         // was just created in this function call - it can never equal
         // any real session's alias, so this lookup was guaranteed to
         // return NULL every time, and the result was dereferenced
         // unconditionally below. Any client sending a bare "invite:"
         // crashed the entire process, not just its own connection -
         // same class of bug as the two GuildManager crashes fixed
         // this session. This generic ServerCore layer has no
         // name-based session lookup available (that only exists as a
         // Zephora-specific addition on top of TCPServer), so a real
         // fix for "find a session by the player name they typed"
         // isn't possible at this layer without a larger change; this
         // stops the crash without pretending to fix that gap.
         // getSessionByAlias() now returns an already-pin()'d
         // shared_ptr - see its own doc comment - so tempSession itself
         // guarantees the session stays alive for this write() call
         // below, no separate pin() needed.
         auto tempSession = session.server->getSessionByAlias(&invitee);
         if (!tempSession)
         {
            coreutils::Log(coreutils::LOG_DEBUG_1) << "invite: could not find session for " << invitee;
            return 0;
         }
         std::stringstream temp;
         temp << "invite:" << request[1] << ":" << invitee;
         tempSession->write(temp.str());
         return 1;
      }

      // Was subscriptions[request[1].str()] - map::operator[] default-
      // constructs (here: NULL) and INSERTS an entry for any id that
      // isn't already registered, and subscription->id below was then
      // dereferenced completely unconditionally. Any subscribe:/
      // unsubscribe:/event: naming an id nobody ever created (confirmed
      // live: "subscribe:invite", sent by the client on every scene
      // load, referencing an "invite" subscription that plain doesn't
      // exist anywhere - real invites are Publish()'d under a
      // per-group-guid id, not a shared "invite" channel) null-derefs
      // and takes down the ENTIRE process, every connection included,
      // not just the one that sent it. find() avoids polluting the map
      // with the bogus NULL entry too.
      auto it = subscriptions.find(request[1].str());
      if (it == subscriptions.end())
      {
         coreutils::Log(coreutils::LOG_DEBUG_1) << "Unknown subscription id: " << request[1].str();
         return 0;
      }
      auto subscription = it->second;

      if (request[1].equals(subscription->id))
      {
         if (request[0] == "unpublish")
         {
            subscriptions.erase(request[1].str());
         }
         else if (request[0] == "subscribe")
         {
            subscription->subscribe(session);
            return 1;
         }
         else if (request[0] == "unsubscribe")
         {
            subscription->unsubscribe(session);
            return 1;
         }
         else if (request[0] == "event")
         {
            std::stringstream out;
            subscription->process(request, out, session);
            if (subscription->mode == "*ANYONE")
            {
               subscription->event(out);
               return 1;
            }
            else if (subscription->mode == "*SUBSCRIBERS")
            {
               if (subscription->ifSubscriber(session))
               {
                  subscription->event(out);
                  return 1;
               }
            }
         }
         return 0;
      }
      return 0;
   }
}