BedrockServer.cpp

// Manages connections to a single instance of the bedrock server.
#include "BedrockServer.h"
#include "BedrockCommand.h"
#include "sqlitecluster/SQLiteNode.h"

#include <arpa/inet.h>
#include <chrono>
#include <csignal>
#include <cstring>
#include <fstream>
#include <future>
#include <sys/resource.h>
#include <sys/time.h>

#include <bedrockVersion.h>
#include <BedrockCore.h>
#include <BedrockPlugin.h>
#include <libstuff/libstuff.h>
#include <libstuff/SRandom.h>
#include <libstuff/AutoTimer.h>
#include <ConflictLockGuard.h>
#include <sqlitecluster/SQLitePeer.h>

thread_local atomic<SQLiteNodeState> BedrockServer::_nodeStateSnapshot = SQLiteNodeState::UNKNOWN;

// Temporary object we're using to log typical timing for various stages of shutting down.
struct
{
    chrono::time_point<chrono::steady_clock> shutdownStart{chrono::steady_clock::now()};
    chrono::time_point<chrono::steady_clock> safeNodeState{chrono::steady_clock::now()};
    chrono::time_point<chrono::steady_clock> commandsComplete{chrono::steady_clock::now()};
    chrono::time_point<chrono::steady_clock> serverDestructor{chrono::steady_clock::now()};
} shutdownTimer;

void BedrockServer::syncWrapper()
{
    // Initialize the thread.
    SInitialize(_syncThreadName);
    isSyncThread = true;

    while (true) {
        // If the server's set to be detached, we wait until that flag is unset, and then start the sync thread.
        if (_detach) {
            // If we're set detached, we assume we'll be re-attached eventually, and then be `RUNNING`.
            SINFO("Bedrock server entering detached state.");
            _shutdownState.store(RUNNING);

            // Detach any plugins now
            for (auto plugin : plugins) {
                plugin.second->onDetach();
            }
            _pluginsDetached = true;
            while (_detach) {
                if (shutdownWhileDetached) {
                    SINFO("Bedrock server exiting from detached state.");
                    return;
                }
                // Just wait until we're attached.
                SINFO("Bedrock server sleeping in detached state.");
                sleep(1);
            }
            SINFO("Bedrock server entering attached state.");
            _resetServer();
        }
        sync();

        // Now that we've run the sync thread, we can exit if it hasn't set _detach again.
        if (!_detach) {
            break;
        }
    }

    // Break out of `poll` in main.cpp.
    _notifyDone.push(true);
    SINFO("Exiting syncWrapper");
}

shared_ptr<SQLitePool> BedrockServer::getDBPool()
{
    return atomic_load(&_dbPool);
}

void BedrockServer::sync()
{
    // Parse out the number of worker threads we'll use. The DB needs to know this because it will expect a
    // corresponding number of journal tables.
    int workerThreads = args.calc("-workerThreads");

    // If still no value, use the number of cores on the machine, if available.
    SINFO("Note: thread::hardware_concurrency() is: " << thread::hardware_concurrency());
    workerThreads = workerThreads ? workerThreads : max(1u, thread::hardware_concurrency());

    // A minimum of *2* worker threads are required. One for blocking writes, one for other commands.
    if (workerThreads < 2) {
        workerThreads = 2;
    }

    size_t journalTables = workerThreads;
    if (args.isSet("-journalTables")) {
        journalTables = args.calcU64("-journalTables");
    }

    // Initialize the DB.
    int64_t mmapSizeGB = args.isSet("-mmapSizeGB") ? stoll(args["-mmapSizeGB"]) : 0;

    // We use fewer FDs on test machines that have other resource restrictions in place.

    SQLite::journalZstdDictionaryID = args.calc("-journalZstdDictionaryID");
    if (args.isSet("-journalDeleterBatchSize")) {
        const int64_t journalDeleterBatchSize = args.calc64("-journalDeleterBatchSize");
        if (journalDeleterBatchSize >= 0) {
            BedrockJournalDeleter::deleterBatchSize = journalDeleterBatchSize;
        } else {
            SWARN("Ignoring -journalDeleterBatchSize '" << args["-journalDeleterBatchSize"] << "', it can't be negative.");
        }
    }
    vector<function<void()>> callbacks;
    for (const auto& plugin : plugins) {
        if (plugin.second->afterCommitCallback) {
            callbacks.emplace_back(plugin.second->afterCommitCallback);
        }
    }
    callbacks.emplace_back([this]() {
        _journalDeleter.wake();
    });
    auto dbPoolDestroyed = make_shared<promise<void>>();
    auto dbPoolDestructionComplete = dbPoolDestroyed->get_future();
    SINFO("Setting dbPool size to: " << _dbPoolSize);
    atomic_store(&_dbPool, shared_ptr<SQLitePool>(
        new SQLitePool(_dbPoolSize, args["-db"], args.calc("-cacheSize"), args.calc("-maxJournalSize"), journalTables, mmapSizeGB, args.isSet("-newDBsUseHctree"), args["-checkpointMode"], callbacks),
        [dbPoolDestroyed](SQLitePool* pool) {
        delete pool;
        dbPoolDestroyed->set_value();
        }));
    SQLite& db = getDBPool()->getBase();
    _journalDeleter.start();

    // Allow plugins to read from the DB at startup.
    for (auto plugin : plugins) {
        plugin.second->initializeFromDB(db);
    }

    // And the sync node.
    uint64_t firstTimeout = STIME_US_PER_M * 2 + SRandom::rand64() % STIME_US_PER_S * 30;

    // Initialize the shared pointer to our sync node object.
    atomic_store(&_syncNode, make_shared<SQLiteNode>(*this, getDBPool(), args["-nodeName"], args["-nodeHost"],
                                                            args["-peerList"], _configuredPriority, _maxOutstandingWALFrames,
                                                            firstTimeout, _version, args["-commandPortPrivate"]));
    _syncNode->registerMessageHandler("CRASH_COMMAND", [this](const SData& message) {
        _handleCrashCommand(message);
    });

    atomic_store(&_clusterMessenger, make_shared<SQLiteClusterMessenger>(_syncNode));

    // The node is now coming up, and should eventually end up in a `LEADING` or `FOLLOWING` state. We can start adding
    // our worker threads now. We don't wait until the node is `LEADING` or `FOLLOWING`, as it's state can change while
    // it's running, and our workers will have to maintain awareness of that state anyway.
    SINFO("Starting " << workerThreads << " worker threads.");
    list<thread> workerThreadList;
    for (int threadId = 0; threadId < workerThreads; threadId++) {
        workerThreadList.emplace_back(&BedrockServer::worker, this, threadId);
    }

    // Now we jump into our main loop.
    uint64_t nextActivity = STimeNow();

    // The upgrade command finishes asynchronously, so the sync loop finds out how it went here, from the command's
    // destructor. `upgradeCommand` is only dereferenced inside this callback, which runs while the command is still
    // alive. `~BedrockCommand` can run on a worker thread, so the flag it clears is atomic.
    BedrockCommand* upgradeCommand = nullptr;
    atomic<bool> upgradeCommandRunning(false);
    function<void()> upgradeCommandFinished = [&]() {
        if (upgradeCommand->complete && upgradeCommand->response.methodLine == "200 OK") {
            _upgradeCompleted = true;
            SINFO("UpgradeDB succeeded, done.");
            _notifyDone.push(true);
        } else {
            SINFO("UpgradeDB failed, trying again.");
        }
        upgradeCommandRunning = false;
    };

    // Timer for S_poll performance logging. Created outside the loop because it's cumulative.
    AutoTimer pollTimer("sync thread poll");
    AutoTimer postPollTimer("sync thread PostPoll");

    // Set the function that will be called if this thread's signal handler catches an unrecoverable error, like a
    // segfault.
    SSetSignalHandlerDieFunc([]() {
        return "Crashing in sync thread.";
    });

    do
    {
        // If there were commands waiting on our commit count to come up-to-date, we'll move them back to the main
        // command queue here. There's no place in particular that's best to do this, so we do it at the top of this
        // main loop, as that prevents it from ever getting skipped in the event that we `continue` early from a loop
        // iteration.
        // We also move all commands back to the main queue here if we're shutting down, just to make sure they don't
        // end up lost in the ether.
        {
            SAUTOLOCK(_futureCommitCommandMutex);

            // First, see if anything has timed out, and move that back to the main queue.
            if (_futureCommitCommandTimeouts.size()) {
                uint64_t now = STimeNow();
                auto it = _futureCommitCommandTimeouts.begin();
                while (it != _futureCommitCommandTimeouts.end() && it->first < now) {
                    // Find commands depending on this commit.
                    auto itPair = _futureCommitCommands.equal_range(it->second);
                    for (auto cmdIt = itPair.first; cmdIt != itPair.second; cmdIt++) {
                        // Check for one with this timeout.
                        if (cmdIt->second->timeout() == it->first) {
                            // This command has the right commit count *and* timeout, return it.
                            SINFO("Returning command (" << cmdIt->second->request.methodLine << ") waiting on commit " << cmdIt->first
                                  << " to queue, timed out at: " << now << ", timeout was: " << it->first << ".");

                            // Goes back to the main queue, where it will hit it's timeout in a worker thread.
                            _commandQueue.push(move(cmdIt->second));

                            // And delete it, it's gone.
                            _futureCommitCommands.erase(cmdIt);

                            // Done.
                            break;
                        }
                    }
                    it++;
                }

                // And remove everything we just iterated through.
                if (it != _futureCommitCommandTimeouts.begin()) {
                    _futureCommitCommandTimeouts.erase(_futureCommitCommandTimeouts.begin(), it);
                }
            }

            // Anything that hasn't timed out might be ready to return because the commit count is up-to-date.
            if (!_futureCommitCommands.empty()) {
                uint64_t commitCount = db.getCommitCount();
                auto it = _futureCommitCommands.begin();
                while (it != _futureCommitCommands.end() && (it->first <= commitCount || _shutdownState.load() != RUNNING)) {
                    // Save the timeout since we'll be moving the command, thus making this inaccessible.
                    uint64_t commandTimeout = it->second->timeout();
                    SINFO("Returning command (" << it->second->request.methodLine << ") waiting on commit " << it->first
                          << " to queue, now have commit " << commitCount);
                    _commandQueue.push(move(it->second));

                    // Remove it from the timed out list as well.
                    auto itPair = _futureCommitCommandTimeouts.equal_range(commandTimeout);
                    for (auto timeoutIt = itPair.first; timeoutIt != itPair.second; timeoutIt++) {
                        if (timeoutIt->second == it->first) {
                            _futureCommitCommandTimeouts.erase(timeoutIt);
                            break;
                        }
                    }
                    it++;
                }
                if (it != _futureCommitCommands.begin()) {
                    _futureCommitCommands.erase(_futureCommitCommands.begin(), it);
                }
            }
        }

        // If we're in a state where we can initialize shutdown, then go ahead and do so.
        // Having responded to all clients means there are no *local* clients, but it doesn't mean there are no
        // escalated commands. This is fine though - if we're following, there can't be any escalated commands, and if
        // we're leading, then the next update() loop will set us to standing down, and then we won't accept any new
        // commands, and we'll shortly run through the existing queue.
        if (_shutdownState.load() == COMMANDS_FINISHED) {
            SINFO("All clients responded to, " << BedrockCommand::getCommandCount() << " commands remaining.");
            if (_syncNode->beginShutdown()) {
                SINFO("Beginning shuttdown of sync node.");
                // This will cause us to skip the next `poll` iteration which avoids a 1 second wait.
                _notifyDoneSync.push(true);
            }
        }

        // The fd_map contains a list of all file descriptors (eg, sockets, Unix pipes) that poll will wait on for
        // activity. Once any of them has activity (or the timeout ends), poll will return.
        fd_map fdm;

        // Pre-process any sockets the sync node is managing (i.e., communication with peer nodes).
        _notifyDoneSync.prePoll(fdm);
        _syncNode->prePoll(fdm);

        // Wait for activity on any of those FDs, up to a timeout.
        const uint64_t now = STimeNow();
        {
            AutoTimerTime pollTime(pollTimer);
            S_poll(fdm, max(nextActivity, now) - now);
            if (SCheckSignal(SIGTERM)) {
                _notifyDone.push(true);
            }
        }

        // And set our next timeout for 1 second from now.
        nextActivity = STimeNow() + STIME_US_PER_S;

        // `preUpdateState` is the synchronization state (i.e., FOLLOWING, LEADING) of the node before allowing it to change.
        // We need to keep track of when this changes such that we can update our own state here (i.e., if we fall out of
        // leading, we will need to notice that and realize we can't attempt to to write to the database).
        // The name of this variable is pre-update in the sense of `_syncNode->update()` which is called below, but we do not
        // set it just before that call, because it is now also possible for the state to change in postPoll -- for example,
        // if we lose connection to a peer, and that peer was the last one that gave us quorum, then we can drop from
        // LEADING to SEARCHING. This transition triggers our "fall out of leading" logic.
        SQLiteNodeState preUpdateState = _syncNode->getState();

        // Process any network traffic that happened. Scope this so that we can change the log prefix and have it
        // auto-revert when we're finished.
        {
            // Set the default log prefix.
            SAUTOPREFIX(SData{});

            // Process any activity in our plugins.
            AutoTimerTime postPollTime(postPollTimer);
            _syncNode->postPoll(fdm, nextActivity);
            _notifyDoneSync.postPoll(fdm);
        }

        // Ok, let the sync node update for as many iterations as it requires. We'll update the replication
        // state when it's finished.
        while (_syncNode->update()) {
        }
        _leaderVersion.store(_syncNode->getLeaderVersion());

        // If we're not leading, move any commands from the blocking queue back to the main queue.
        if (getState() != SQLiteNodeState::LEADING && getState() != preUpdateState) {
            auto commands = _blockingCommandQueue.getAll();
            SINFO("Moving " << commands.size() << " commands from blocking queue to main queue.");
            for (auto& cmd : commands) {
                _commandQueue.push(move(cmd));
            }

            // Clear rate limit state since we're transitioning away from leading. Any blocks accumulated
            // while leading are no longer meaningful — commands will be re-queued against new leader state.
            _blockingCommandQueue.clearRateLimits();
        }

        // Reset some state on falling out of leading. This could be normal, if we're just shutting down, or it
        // could be exceptional, in the case we've lost connection to the cluster or some other problem.
        if ((preUpdateState == SQLiteNodeState::LEADING || preUpdateState == SQLiteNodeState::STANDINGDOWN) &&
            (getState() != SQLiteNodeState::LEADING && getState() != SQLiteNodeState::STANDINGDOWN)) {
            shutdownTimer.safeNodeState = chrono::steady_clock::now();
        }

        // Now that we've cleared any state associated with switching away from leading, we can bail out and try again
        // until we're either leading or following.
        if (getState() != SQLiteNodeState::LEADING && getState() != SQLiteNodeState::FOLLOWING && getState() != SQLiteNodeState::STANDINGDOWN) {
            continue;
        }

        // If _upgradeCompleted is set, I.e., we've run at least one upgrade, we can skip this block.
        // Otherwise, if we're leading, run the upgrade command, and wait for it to finish before we run anything else.
        if (!_upgradeCompleted && !upgradeCommandRunning && getState() == SQLiteNodeState::LEADING) {
            if (!_syncNode->majorityOfFollowersSubscribed()) {
                // We stand up as soon as a majority of peers are logged in, but a peer doesn't receive transactions
                // until it has gone SUBSCRIBING -> FOLLOWING, which takes a moment longer. Committing the upgrade in
                // that window means most of the cluster doesn't get the schema change from us and has to pick it up
                // by synchronizing later, so wait for them instead.
                SINFO("Waiting for followers to subscribe before running UpgradeDB.");
                continue;
            }
            auto upgrade = make_unique<BedrockServerUpgradeCommand>(*this);
            upgradeCommand = upgrade.get();
            upgrade->destructionCallback = &upgradeCommandFinished;
            upgradeCommandRunning = true;
            runCommand(move(upgrade), true);

            // This interrupts the next poll loop immediately. This prevents a 1-second wait when running as a single server.
            _notifyDoneSync.push(true);
            continue;
        }
    } while (!_syncNode->shutdownComplete() || BedrockCommand::getCommandCount());

    SSetSignalHandlerDieFunc([]() {
        return "Dying in shutdown";
    });

    // We've finished shutting down the sync node, tell the workers that it's finished.
    _shutdownState.store(DONE);
    SINFO("Sync thread finished with commands.");

    // We just fell out of the loop where we were waiting for shutdown to complete. Update the state one last time when
    // the writing replication thread exits.
    if (getState() > SQLiteNodeState::WAITING) {
        // This should no longer be possible with fast shutdown.
        SWARN("Sync thread exiting in state " << SQLiteNode::stateName(getState()) << ".");
    }

    // Wait for the worker threads to finish.
    int threadId = 0;
    for (auto& workerThread : workerThreadList) {
        SINFO("Joining worker thread '" << "worker" << threadId << "'");
        threadId++;
        workerThread.join();
    }

    // If there's anything left in the command queue here, we'll discard it, because we have no way of processing it.
    if (_commandQueue.size()) {
        SWARN("Sync thread shut down with " << _commandQueue.size() << " queued commands. Commands were: "
              << SComposeList(_commandQueue.getRequestMethodLines()) << ". Clearing.");
        _commandQueue.clear();
    }

    // Same for the blocking queue.
    if (_blockingCommandQueue.size()) {
        SWARN("Sync thread shut down with " << _blockingCommandQueue.size() << " blocking queued commands. Commands were: "
              << SComposeList(_blockingCommandQueue.getRequestMethodLines()) << ". Clearing.");
        _blockingCommandQueue.clear();
    }

    for (auto plugin : plugins) {
        plugin.second->serverStopping();
    }

    _journalDeleter.stop();

    // Release the server's references in dependency order. Other threads may still hold temporary references.
    atomic_store(&_clusterMessenger, shared_ptr<SQLiteClusterMessenger>(nullptr));
    atomic_store(&_syncNode, shared_ptr<SQLiteNode>(nullptr));
    atomic_store(&_dbPool, shared_ptr<SQLitePool>(nullptr));

    // The custom deleter signals only after the pool and all its database handles have been destroyed.
    SINFO("Waiting for DB pool destruction.");
    while (dbPoolDestructionComplete.wait_for(chrono::seconds(5)) != future_status::ready) {
        SWARN("Still waiting for DB pool destruction.");
    }
    SINFO("DB pool destroyed.");

    // If we're not detaching, save that we're shutting down.
    if (!_detach) {
        ofstream file("/var/log/bedrock_shutdown", ios::app);
        if (file) {
            file << "shutdown " << getpid() << " " << SComposeTime("%Y-%m-%dT%H:%M:%S", STimeNow()) << endl;
            file.close();
        }
    }

    // We're really done, store our flag so main() can be aware.
    _syncLoopShouldBeRunning.store(false);
}

void BedrockServer::worker(int threadId)
{
    // Worker 0 is the "blockingCommit" thread.
    SInitialize(threadId ? "worker" + to_string(threadId) : "blockingCommit");
    isBlockingCommitThread = (threadId == 0);

    // Command to work on. This default command is replaced when we find work to do.
    unique_ptr<BedrockCommand> command(nullptr);

    // Which command queue do we use? The blockingCommit thread special and does blocking commits from the blocking queue.
    BedrockCommandQueue& commandQueue = threadId ? _commandQueue : _blockingCommandQueue;

    // We just run this loop looking for commands to process forever. There's a check for appropriate exit conditions
    // at the bottom, which will cause our loop and thus this thread to exit when that becomes true.
    while (true) {
        try {
            // Set a signal handler function that we can call even if we die early with no command.
            SSetSignalHandlerDieFunc([&]() {
                return "Die function called early with no command, probably died in `commandQueue.get`.";
            });

            // Get the next one.
            command = commandQueue.get(100000);

            SAUTOPREFIX(command->request);
            SINFO("Dequeued command " << command->request.methodLine << " (" << command->id << ") in worker, "
                  << commandQueue.size() << " commands in " << (threadId ? "" : "blocking") << " queue.");

            // Capture the identifier and command name so we can attribute the blocking execution time back to the
            // rate limiter after the command finishes. We time every command run on the blocking thread, recording against
            // the global rate limiter and its command name always, and its identifier when it is set.
            const string blockingIdentifier = (threadId == 0) ? command->blockingQueueRateLimitIdentifier : "";
            const string commandName = command->request.methodLine;
            const uint64_t blockingStart = (threadId == 0) ? STimeNow() : 0;
            SQLite::resetCommitLockWait();

            runCommand(move(command), threadId == 0, false);

            if (blockingStart) {
                // Waiting for the commit lock is caused by unrelated cluster contention, so it isn't the identifier's
                // work and must not count toward its rate limit.
                const uint64_t elapsedUS = STimeNow() - blockingStart;
                const uint64_t commitLockWaitUS = SQLite::getCommitLockWait();
                _blockingCommandQueue.recordExecutionTime(blockingIdentifier, commandName, elapsedUS > commitLockWaitUS ? elapsedUS - commitLockWaitUS : 0);
            }
        } catch (const BedrockCommandQueue::timeout_error& e) {
            // No commands to process after 1 second.
            // If the sync node has shut down, we can return now, there will be no more work to do.
            if (_shutdownState.load() == DONE) {
                SINFO("No commands found in queue and DONE.");
                return;
            }
        }
    }
}

bool BedrockServer::isShuttingDown()
{
    bool shuttingDown = false;
    auto _syncNodeCopy = atomic_load(&_syncNode);
    if (_shutdownState.load() != RUNNING || (_syncNodeCopy && _syncNodeCopy->getState() == SQLiteNodeState::STANDINGDOWN)) {
        shuttingDown = true;
    }
    return shuttingDown;
}

void BedrockServer::runCommand(unique_ptr<BedrockCommand>&& _command, bool isBlocking, bool hasDedicatedThread)
{
    // This takes ownership of the passed command. By calling the move constructor, the caller's unique_ptr is now empty, and so when the one here goes out of scope (i.e., this function
    // returns), the command is destroyed.
    unique_ptr<BedrockCommand> command(move(_command));

    SAUTOPREFIX(command->request);

    // If the command has already timed out, reply as such.
    if (BedrockCore::isTimedOut(command, nullptr, this)) {
        _reply(command);
        return;
    }

    // Set the function that lets the signal handler know which command caused a problem, in case that happens.
    // If a signal is caught on this thread, which should only happen for unrecoverable, yet synchronous
    // signals, like SIGSEGV, this function will be called.
    SSetSignalHandlerDieFunc([&]() {
        atomic_load(&_clusterMessenger)->runOnAll(_generateCrashMessage(command));
        return addLogParams("CRASHING from BedrockServer::runCommand, command:" + command->request.methodLine, command->request.nameValueMap);
    });

    // If we dequeue a status or control command, handle it immediately.
    if (_handleIfStatusOrControlCommand(command)) {
        return;
    }

    // Check if this command would be likely to cause a crash
    if (_wouldCrash(command)) {
        // If so, make a lot of noise, and respond 500 without processing it. WARN instead of ALERT to avoid BugBot issues.
        SWARN("REJECTING CRASH-INDUCING COMMAND, command:" + command->request.methodLine, command->request.nameValueMap);
        command->response.methodLine = "500 Refused";
        command->complete = true;
        _reply(command);
        return;
    }

    // If there's no sync node (because we're detaching/attaching), we can't run a command right now.
    // Also, if this command is scheduled in the future, we can't run it either, we need to enqueue it to run at that point.
    // This functionality will go away as we remove the queues from bedrock, and so this can be removed at that time.
    {
        auto _syncNodeCopy = atomic_load(&_syncNode);
        if (!_syncNodeCopy || command->request.calcU64("commandExecuteTime") > STimeNow()) {
            _commandQueue.push(move(command));
            return;
        }
    }

    size_t waitCount = 0;

    // We just spin until the node looks ready to go. Typically, this doesn't happen expect briefly at startup.
    // We specifically exclude `STANDINGDOWN` because even though we *can* process commands in this state, the intention
    // here is to shut the node down and so we want to stop building a backlog of in-process commands that prevent
    // standdown from completing. They will get blocked until we hit our final FOLLOWING state and then finished.
    while (!dbReadyToHandleRequests() || getState() == SQLiteNodeState::STANDINGDOWN) {
        if (dynamic_cast<BedrockServerUpgradeCommand*>(command.get())) {
            // the upgrade command is exempted from this loop because it needs to run to make the
            // dbready to handle commands.
            break;
        }
        // It's feasible that our command times out in this loop. In this case, we do not have a DB object to pass.
        // The only implication of this is the response does not get the commitCount attached to it.
        if (BedrockCore::isTimedOut(command, nullptr, this)) {
            _reply(command);
            return;
        }

        // This sleep call is pretty ugly, but it should almost never happen. We're accepting the potential
        // looping sleep call for the general case where we just check some bools and continue, instead of
        // avoiding the sleep call but having every thread lock a mutex here on every loop.
        usleep(10000);
        waitCount++;
    }
    if (waitCount) {
        SINFO("Waited for " << waitCount << " loops for node to be ready.");
    }

    // If we're following, we will automatically escalate any command that's:
    // 1. Not already complete (complete commands are likely already returned from leader with legacy escalation)
    // and is marked as `escalateImmediately` (which lets them skip the queue, which is particularly useful if they're waiting
    // for a previous commit to be delivered to this follower);
    // 2. Any commands if the current version of the code is not the same one as leader is executing.
    if (getState() == SQLiteNodeState::FOLLOWING && !command->complete && (command->escalateImmediately || _version != _leaderVersion.load())) {
        auto _clusterMessengerCopy = atomic_load(&_clusterMessenger);
        if (command->escalateImmediately && _clusterMessengerCopy && _clusterMessengerCopy->runOnPeer(*command, true)) {
            // command->complete is now true for this command. It will get handled a few lines below.
            SINFO("Immediately escalated " << command->request.methodLine << " to leader.");
        } else if (_version != _leaderVersion.load() && _clusterMessengerCopy && _clusterMessengerCopy->runOnPeer(*command, false)) {
            SINFO("Escalated " << command->request.methodLine << " to follower peer.");
        } else {
            SINFO("Couldn't escalate command " << command->request.methodLine << " to " << (command->escalateImmediately ? "leader" : "follower peer") << ", queuing it again.");
            _commandQueue.push(move(command));
            return;
        }
    }

    // If we happen to be synchronizing but the command port is open, which is an uncommon but possible scenario (i.e., we were momentarily disconnected from leader and need to catch back
    // up), we will forward commands to any other follower similar to if we were running as a different version from leader.
    if (getState() == SQLiteNodeState::SYNCHRONIZING) {
        auto _clusterMessengerCopy = atomic_load(&_clusterMessenger);
        bool result = _clusterMessengerCopy->runOnPeer(*command, false);
        if (result) {
            SINFO("Synchronizing while accepting commands; successfully forwarded the command to peer", {{"command", command->request.methodLine}});
        } else {
            SHMMM("Synchronizing while accepting commands, but failed to forward the command to peer.", {{"command", command->request.methodLine}});
        }
    }

    // If this command is already complete, then we should be a follower, and the sync node got a response back
    // from a command that had been escalated to leader, and queued it for a worker to respond to. We'll send
    // that response now.
    if (command->complete) {
        // If this command is already complete, we can return it to the caller.
        // Make sure we have an initiatingClientID at this point. If we do, but it's negative, it's for a
        // client that we can't respond to, so we don't bother sending the response.
        SASSERT(command->initiatingClientID);
        if (command->initiatingClientID > 0) {
            _reply(command);
        }

        // This command is done, move on to the next one.
        return;
    }

    if (command->request.isSet("mockRequest")) {
        SINFO("mockRequest set for command '" << command->request.methodLine << "'.");
    }

    // Commands named in `-synchronousCommands` run on the blocking commit thread. Hand this one off to that queue,
    // unless we're already on that thread (worker 0 calls us with `isBlocking` set, and re-pushing here would spin).
    // If we're not LEADING we take the normal path instead, which peeks locally and escalates to leader if it needs to
    // write; the leader sets this flag again when it builds the escalated command.
    if (command->isSynchronous && !isBlocking && getState() == SQLiteNodeState::LEADING) {
        SINFO("Sending synchronous command '" << command->request.methodLine << "' to blocking queue with size "
              << _blockingCommandQueue.size());
        try {
            // BedrockBlockingCommandQueue::push() guarantees that any exception (e.g., rate limiting) is thrown before
            // the underlying move() into the queue occurs, so `command` is still valid in the catch block.
            _blockingCommandQueue.push(move(command));
        } catch (const SException& e) {
            command->response.methodLine = e.what();
            command->complete = true;
            _reply(command);
        }
        return;
    }

    int64_t lastConflictIdentifier = 0;
    string lastConflictLocation;
    while (true) {
        // If there are outstanding HTTPS requests on this command (from a previous call to `peek`) we process them here.
        command->waitForHTTPSRequests();

        // Get a DB handle to work on. This will automatically be returned when dbScope goes out of scope.
        auto dbPool = getDBPool();
        if (!dbPool) {
            SERROR("Can't run a command with no DB pool");
        }
        {
            uint64_t dbHandleStartTime = STimeNow();
            SQLiteScopedHandle dbScope(*dbPool, dbPool->getIndex());
            SQLite& db = dbScope.db();
            command->recordTiming(BedrockCommand::DB_HANDLE, dbHandleStartTime);
            BedrockCore core(db, *this);

            // If the command has already timed out when we get it, we can return early here without peeking it.
            // We'd also catch that the command timed out in `peek`, but this can cause some weird side-effects: a
            // command that already has a HTTPS request attached doesn't get peeked, so anything that re-queues such a
            // command without checking the timeout first can bounce it between queues indefinitely.
            if (core.isTimedOut(command, &db, this)) {
                _reply(command);
                return;
            }

            // If this command is dependent on a commitCount newer than what we have (maybe it's a follow-up to a
            // command that was escalated to leader), we'll set it aside for later processing. When the sync node
            // finishes its update loop, it will re-queue any of these commands that are no longer blocked on our
            // updated commit count.
            uint64_t commitCount = db.getCommitCount();
            uint64_t commandCommitCount = command->request.calcU64("commitCount");
            if (commandCommitCount > commitCount) {
                SAUTOLOCK(_futureCommitCommandMutex);
                auto newQueueSize = _futureCommitCommands.size() + 1;
                SINFO("Command (" << command->request.methodLine << ") depends on future commit (" << commandCommitCount
                      << "), Currently at: " << commitCount << ", storing for later. Queue size: " << newQueueSize);
                _futureCommitCommandTimeouts.insert(make_pair(command->timeout(), commandCommitCount));
                _futureCommitCommands.insert(make_pair(commandCommitCount, move(command)));

                // Don't count this as `in progress`, it's just sitting there.
                if (newQueueSize > 100) {
                    SHMMM("_futureCommitCommands.size() == " << newQueueSize);
                }
                return;
            }

            // If the command should run prePeek, do that now .
            if (!command->repeek && !command->httpsRequests.size() && command->shouldPrePeek()) {
                core.prePeekCommand(command, isBlocking);

                if (command->complete) {
                    _reply(command);
                    break;
                }
            }

            auto* timer = new BedrockCore::AutoTimer(command, BedrockCommand::QUEUE_PAGE_LOCK);
            uint64_t conflictLockStartTime = 0;
            if (lastConflictIdentifier) {
                conflictLockStartTime = STimeNow();
            }
            {
                ConflictLockGuard conflictLock(lastConflictIdentifier);
                if (lastConflictIdentifier) {
                    SINFO("Waited " << (STimeNow() - conflictLockStartTime) << "us for lock on conflict key " << lastConflictIdentifier << " (" << lastConflictLocation << ").");
                }
                delete timer;

                // If the command has any httpsRequests from a previous `peek`, we won't peek it again unless the
                // command has specifically asked for that.
                // If peek succeeds, then it's finished, and all we need to do is respond to the command at the bottom.
                bool calledPeek = false;
                BedrockCore::RESULT peekResult = BedrockCore::RESULT::INVALID;
                if (command->repeek || !command->httpsRequests.size()) {
                    peekResult = core.peekCommand(command, isBlocking);
                    calledPeek = true;
                }

                if (!calledPeek || peekResult == BedrockCore::RESULT::SHOULD_PROCESS) {
                    // We've just unsuccessfully peeked a command, which means we're in a state where we might want to
                    // write it. We'll flag that here, to keep the node from falling out of LEADING/STANDINGDOWN
                    // until we're finished with this command.
                    if (command->httpsRequests.size()) {
                        if (command->repeek || !command->areHttpsRequestsComplete()) {
                            // Roll back the existing transaction, but only if we are inside an transaction
                            if (calledPeek) {
                                core.rollback(command->getMethodName());
                            }

                            // Jump back to the top of our main `while (true)` loop and run the network activity loop again.
                            continue;
                        }
                    }

                    // We checked this before peeking, but we can have started leading since then, which puts us back to
                    // waiting on the DB upgrade. Send the command back to the main queue, where it waits for that like
                    // any other command. The command that runs the upgrade is exempt, or it would wait on itself.
                    if (!dbReadyToHandleRequests() && !dynamic_cast<BedrockServerUpgradeCommand*>(command.get())) {
                        core.rollback(command->getMethodName());
                        dbScope.release();
                        SINFO("Started leading before the DB upgrade finished, re-queueing " << command->request.methodLine << ".");
                        _commandQueue.push(move(command));
                        return;
                    }

                    // Peek wasn't enough to handle this command, so it needs to write, and only the leader can do
                    // that.
                    if (getState() != SQLiteNodeState::LEADING) {
                        // Roll back the transaction, it'll get re-run wherever this command ends up.
                        core.rollback(command->getMethodName());
                        dbScope.release();
                        auto _clusterMessengerCopy = atomic_load(&_clusterMessenger);
                        if (_clusterMessengerCopy && _clusterMessengerCopy->runOnPeer(*command, true)) {
                            SINFO("Escalated " << command->request.methodLine << " to leader and complete, responding.");
                            _reply(command);
                        } else {
                            // TODO: Something less naive that considers how these failures happen rather than a simple
                            // endless loop of requeue and retry.
                            SINFO("Couldn't escalate command " << command->request.methodLine << " to leader. We are in state: " << SQLiteNode::stateName(getState()));
                            _commandQueue.push(move(command));
                        }

                        // Done with this command, look for the next one.
                        break;
                    }

                    // In this case, there's nothing blocking us from processing this in a worker, so let's try it.
                    BedrockCore::RESULT result = core.processCommand(command, isBlocking);
                    if (result == BedrockCore::RESULT::NEEDS_COMMIT) {
                        // If processCommand returned true, then we need to do a commit. Otherwise, the command is
                        // done, and we just need to respond. `commit` takes the commit lock, so we serialize against
                        // every other thread that's committing. We can still conflict with a transaction that another
                        // thread committed between our call to `processCommand` and our call to `commit`. This is
                        // scoped to the minimum time required.
                        SQLiteCore::COMMIT_RESULT commitResult = SQLiteCore::COMMIT_RESULT::CONFLICT;
                        uint64_t transactionID = 0;
                        string transactionHash;

                        // Let's use a default value of 0 for the time to acquire the lock, which would work as a try_lock.
                        chrono::microseconds timeToCommit = chrono::microseconds(0);
                        {
                            BedrockCore::AutoTimer timer(command, isBlocking ? BedrockCommand::BLOCKING_COMMIT_WORKER : BedrockCommand::COMMIT_WORKER);
                            void (* onPrepareHandler)(SQLite& db, int64_t tableID) = nullptr;
                            bool enableOnPrepareNotifications = command->shouldEnableOnPrepareNotification(db, &onPrepareHandler);

                            // We want to calculate the remaining time for this command before we try to commit. We cap the wait at the process timeout. It's possible that it will throw.
                            // If that's the case, we will still allow the commit flow to happen passing a 0 timeout, making it behave as a try_lock and still commit it if there's no wait time for the mutex.
                            try {
                                timeToCommit = core.getRemainingTime(command, true);
                            } catch (const SException& e) {
                                SINFO("Command '" << command->getMethodName() << "' timed out before commit.");
                            }
                            commitResult = core.commit(*_syncNode, transactionID, transactionHash, command->getMethodName(), enableOnPrepareNotifications, onPrepareHandler, timeToCommit, &command->shouldAbort);

                            if (getState() != SQLiteNodeState::LEADING) {
                                SINFO("Stopped leading while trying to commit, will retry.");

                                // Jump back to the top of the main loop but skip the check that would push this to the blocking commit queue.
                                // It doesn't make sense to run anything in the blocking queue on a follower. There's a possible edge case where this node
                                // begins leading again before this command runs. In that case, the command will still end up in the blocking commit queue,
                                // possibly with an extra call to `process` over the maximum expected to be allowed. This is very unlikely, though.
                                continue;
                            }
                        }
                        if (commitResult == SQLiteCore::COMMIT_RESULT::OK) {
                            // Tell the sync node that there's been a commit so that it can jump out of it's "poll"
                            // loop and send it to followers. NOTE: we don't check for null here, that should be
                            // impossible inside a worker thread.
                            if (transactionID) {
                                _syncNode->notifyCommit();
                            }
                            _conflictManager.recordTables(command->request.methodLine, db.getTablesUsed());
                            // So we must still be leading, and at this point our commit has succeeded, let's
                            // mark it as complete. We add the currentCommit count here as well.
                            command->response["commitCount"] = to_string(db.getCommitCount());
                            command->complete = true;
                        } else if (commitResult == SQLiteCore::COMMIT_RESULT::TIMEOUT && command->httpsRequests.empty()) {
                            // A long wait for the commit lock means the leader is overloaded. Retrying would run the command
                            // again and queue it behind the same writers, and eventually end up in the blockingCommit thread, so we fail it.
                            // Commands with HTTPS requests keep retrying, because failing them would lose the result of a call
                            // to an outside service that already happened.
                            SINFO("Timed out waiting for commit lock, failing " << command->request.methodLine);
                            command->response.methodLine = "555 Timeout waiting for commit lock";
                            command->response["commitCount"] = to_string(db.getCommitCount());
                            command->complete = true;
                        } else if (!core.isTimedOut(command, &db, this)) {
                            // One of the reasons the commit failed might be because of a timeout. If that's the case,
                            // then we skip this part since everything will be done in the method call above.
                            SINFO("Conflict, abort, or state change committing " << command->request.methodLine);
                            if (_enableConflictPageLocks) {
                                lastConflictLocation = db.getLastConflictLocation();

                                // Journals are always chosen at the time of commit. So in case there was a conflict on the journal in
                                // the previous commit, the chances are very low that we'll choose the same journal, thus, we
                                // don't need to lock our next commit on this page conflict.
                                // Plugins may define other tables on which we should not lock our next commit.
                                if (!SStartsWith(lastConflictLocation, "journal") && (command->getPlugin() == nullptr || command->getPlugin()->shouldLockCommitPageOnConflict(lastConflictLocation))) {
                                    lastConflictIdentifier = db.getLastConflictIdentifier();
                                }
                            }
                        }
                    } else if (result == BedrockCore::RESULT::NO_COMMIT_REQUIRED) {
                        // Nothing to do in this case, `command->complete` will be set and we'll finish as we fall out
                        // of this block.
                    } else if (result == BedrockCore::RESULT::SERVER_NOT_LEADING) {
                        // We won't write regardless.
                        core.rollback(command->getMethodName());

                        // If there are no HTTPS requests, we can just re-queue this command, otherwise, we will
                        // potentially run the same HTTPS requests twice.
                        if (command->httpsRequests.size()) {
                            SALERT("Server stopped leading while running command with HTTPS requests!");
                            command->response.methodLine = "500 Leader stopped leading";
                            _reply(command);
                            break;
                        } else {
                            // Allow for an extra retry and start from the top.
                            SINFO("State changed before 'processCommand' but no HTTPS requests so retrying.");
                        }
                    } else {
                        SERROR("processCommand (" << command->request.getVerb() << ") returned invalid result code: " << (int) result);
                    }
                }
            }
            // If the command was completed above, then we'll go ahead and respond. Otherwise there must have been
            // a conflict or the command was abandoned for a checkpoint, and we'll retry.
            if (command->complete) {
                if (command->shouldPostProcess() && command->response.methodLine == "200 OK") {
                    // PostProcess if the command should run postProcess, and there have been no errors thrown thus far.
                    core.postProcessCommand(command, isBlocking);
                }

                _reply(command);

                // Don't need to retry.
                break;
            }
        }

        // If we're shutting down, or have set a specific max retries, we just try several times in a row and then move the command to the blocking queue.
        int maxRetries = _maxConflictRetries.load();
        if (maxRetries || _shutdownState.load() != RUNNING) {
            if (command->processCount > maxRetries) {
                SINFO("Max retries (" << maxRetries << ") hit in worker, sending '" << command->request.methodLine << "' to blocking queue with size " << _blockingCommandQueue.size());
                try {
                    // BedrockBlockingCommandQueue::push() guarantees that any exception (e.g., rate
                    // limiting) is thrown before the underlying move() into the queue occurs. This
                    // means `command` has not been moved-from when we enter the catch block and the
                    // unique_ptr is still valid for building the error response.
                    _blockingCommandQueue.push(move(command));
                } catch (const SException& e) {
                    command->response.methodLine = e.what();
                    command->complete = true;
                    _reply(command);
                    return;
                }
                return;
            }
        } else {
            // If we're not shutting down, see how long we want to wait until we'll try this command again.
            size_t millisecondsToWait = 0;
            switch (command->processCount) {
                case 1:
                    millisecondsToWait = 10;
                    break;

                case 2:
                    millisecondsToWait = 25;
                    break;

                case 3:
                    millisecondsToWait = 50;
                    break;

                case 4:
                    millisecondsToWait = 100;
                    break;

                case 5:
                    millisecondsToWait = 250;
                    break;

                default:
                    millisecondsToWait = 500;
            }

            // Apply jitter. Take a value that's a whole number up to 50% of ideal time. This allows for adding or subtracting up to 25%.
            millisecondsToWait += ((SRandom::rand64() % (millisecondsToWait / 2)) - (millisecondsToWait / 4));
            SINFO("Waiting " << millisecondsToWait << "ms before retrying command '" << command->request.methodLine << "'.");
            if (hasDedicatedThread) {
                // If we have a dedicated socket thread for this command, we can just sleep here.
                this_thread::sleep_for(chrono::milliseconds(millisecondsToWait));
            } else {
                // Otherwise, re-queue and let another thread try again.
                _commandQueue.push(move(command), STimeNow() + millisecondsToWait * 1000);
                return;
            }
        }
    }
}

bool BedrockServer::_handleIfStatusOrControlCommand(unique_ptr<BedrockCommand>& command)
{
    if (_isStatusCommand(command)) {
        _status(command);
        _reply(command);
        return true;
    } else if (_isControlCommand(command)) {
        // Control commands can only come from localhost (and thus have an empty `_source`)
        // with the exception of non-secure control commands
        if (command->request["_source"].empty() || _isNonSecureControlCommand(command)) {
            _control(command);
        } else {
            SWARN("Got control command " << command->request.methodLine << " on non-localhost socket ("
                  << command->request["_source"] << "). Ignoring.");
            command->response.methodLine = "401 Unauthorized";
        }
        _reply(command);
        return true;
    }
    return false;
}

bool BedrockServer::_wouldCrash(const unique_ptr<BedrockCommand>& command)
{
    // Get a shared lock so that all the workers can look at this map simultaneously.
    shared_lock<decltype(_crashCommandMutex)> lock(_crashCommandMutex);

    // Typically, this map is empty and this returns no results.
    auto commandIt = _crashCommands.find(command->request.methodLine);
    if (commandIt == _crashCommands.end()) {
        return false;
    }

    // If this command crashed with more than one set of identifying values, it means
    // we've already crashed more than one node. Let's fully block this command in that case.
    if (commandIt->second.size() > 1) {
        return true;
    }

    // Look at each crash-inducing command that has the same methodLine.
    for (const STable& values : commandIt->second) {
        // These are all of the keys that need to match to kill this command.
        bool isMatch = true;
        for (auto& pair : values) {
            // We skip Content-Length, as it's added automatically when serializing commands.
            if (SIEquals(pair.first, "Content-Length")) {
                continue;
            }

            // See if our current command even has the blacklisted key.
            auto it = command->request.nameValueMap.find(pair.first);
            if (it == command->request.nameValueMap.end()) {
                // If we didn't find it, the command's not sufficiently similar, and is not blacklisted.
                isMatch = false;
                break;
            }

            // At this point, we must have the same key, but if it doesn't have the same value, then it doesn't match.
            if (it->second != pair.second) {
                isMatch = false;
                break;
            }
        }

        // If we got through the whole list and everything was a match, then this is a match, we think it'll crash.
        if (isMatch) {
            return true;
        }
    }

    // If nothing in our range returned true, then this command looks fine.
    return false;
}

void BedrockServer::_resetServer()
{
    lock_guard<mutex> lock(_portMutex);

    _requestCount = 0;
    if (_commandPortBlockReasons.size()) {
        SWARN("Clearing leftover command port blocks in resetServer (" << _commandPortBlockReasons.size() << " blocks remaining).");
        _commandPortBlockReasons.clear();
    }
    _syncLoopShouldBeRunning = true;
    atomic_store(&_syncNode, shared_ptr<SQLiteNode>(nullptr));
    _shutdownState = RUNNING;
    _shouldBackup = false;
    _commandPortPublic = nullptr;
    _commandPortPrivate = nullptr;
    _pluginsDetached = false;
    _upgradeCompleted = false;
    _shutdownTime = chrono::time_point<chrono::steady_clock>{};
    _blockingCommandQueue.clearRateLimits();

    // Tell any plugins that they can attach now
    for (auto plugin : plugins) {
        plugin.second->onAttach();
    }
}

BedrockServer::BedrockServer(SQLiteNodeState state, const SData& args_)
    : SQLiteServer(), args(args_), _journalDeleter(*this), _syncNode(nullptr), _configuredPriority(args.calc("-priority")), _clusterMessenger(nullptr)
{
}

BedrockServer::BedrockServer(const SData& args_)
    : SQLiteServer(), shutdownWhileDetached(false), args(args_), _journalDeleter(*this), _requestCount(0),
    _isCommandPortLikelyBlocked(false),
    _syncLoopShouldBeRunning(true), _syncNode(nullptr), _configuredPriority(args.calc("-priority")), _clusterMessenger(nullptr), _shutdownState(RUNNING),
    _enableConflictPageLocks(args.test("-enableConflictPageLocks")), _shouldBackup(false), _detach(args.isSet("-bootstrap")),
    _controlPort(nullptr), _commandPortPublic(nullptr), _commandPortPrivate(nullptr), _maxConflictRetries(3),
    _pluginsDetached(false), _socketThreadNumber(0),
    _outstandingSocketThreads(0), _shouldBlockNewSocketThreads(false), _upgradeCompleted(false)
{
    _version = VERSION;

    // This allows the signal thread to notify us when a signal is received to interrupt the current poll loop.
    SSIGNAL_NOTIFY_INTERRUPT = &_notifyDoneSync;

    // Enable the requested plugins, and update our version string if required.
    list<string> pluginNameList = SParseList(args["-plugins"]);
    SINFO("Loading plugins: " << args["-plugins"]);
    vector<string> versions = {_version};
    for (string& pluginName : pluginNameList) {
        auto it = BedrockPlugin::g_registeredPluginList.find(SToUpper(pluginName));
        if (it == BedrockPlugin::g_registeredPluginList.end()) {
            SERROR("Cannot find plugin '" << pluginName << "', aborting.");
        }

        // Create an instance of this plugin.
        BedrockPlugin* plugin = it->second(*this);
        plugins.emplace(make_pair(plugin->getName(), plugin));

        // If the plugin has version info, add it to the list.
        auto info = plugin->getInfo();
        auto iterator = info.find("version");
        if (iterator != info.end()) {
            versions.push_back(plugin->getName() + "_" + iterator->second);
        }
    }
    sort(versions.begin(), versions.end());
    _version = SComposeList(versions, ":");

    list<string> pluginString;
    for (auto& p : plugins) {
        pluginString.emplace_back(p.first);
    }
    SINFO("Creating BedrockServer with plugins: " << SComposeList(pluginString));

    // If `versionOverride` is set, we throw away what we just did and use the overridden value.
    // We'll destruct, sort, and then reconstruct the version string passed in so we aren't relying
    // on the operator to know that they must be sorted.
    if (args.isSet("-versionOverride")) {
        list<string> versionStrings = SParseList(args["-versionOverride"], ':');
        versionStrings.sort();
        _version = SComposeList(versionStrings, ":");
    }

    // Allow enabling tracing at startup.
    if (args.isSet("-enableSQLTracing")) {
        SQLite::enableTrace.store(true);
    }

    // Bypass journald.
    if (args.isSet("-logDirectlyToSyslogSocket")) {
        SSyslogFunc = &SSyslogSocketDirect;
    }

    // Check for commands that will be forced to run on the blocking commit thread.
    if (args.isSet("-synchronousCommands")) {
        list<string> synchronousCommands;
        SParseList(args["-synchronousCommands"], synchronousCommands);
        for (auto& command : synchronousCommands) {
            _synchronousCommands.insert(command);
        }
    }

    // Allow sending control commands when the server's not LEADING/FOLLOWING.
    SINFO("Opening control port on '" << args["-controlPort"] << "'");
    {
        lock_guard<mutex> lock(_portMutex);
        _controlPort = openPort(args["-controlPort"]);
    }

    // If we're bootstraping this node we need to go into detached mode here.
    // The syncWrapper will handle this for us.
    if (_detach) {
        SINFO("Bootstrap flag detected, starting sync node in detach mode.");
    }

    if (args.isSet("-dbPoolSize")) {
        _dbPoolSize = args.calcU64("-dbPoolSize");
    } else {
        _dbPoolSize = args.isSet("-live") ? DEFAULT_POOL_SIZE : 250;
    }

    if (args.isSet("-maxSocketThreads")) {
        _maxSocketThreads = args.calcU64("-maxSocketThreads");
    }

    if (args.isSet("-maxOutstandingWALFrames")) {
        const int64_t maxOutstandingWALFrames = args.calc64("-maxOutstandingWALFrames");
        if (maxOutstandingWALFrames >= 0) {
            _maxOutstandingWALFrames = static_cast<uint64_t>(maxOutstandingWALFrames);
        } else {
            SWARN("Ignoring -maxOutstandingWALFrames '" << args["-maxOutstandingWALFrames"] << "', it can't be negative.");
        }
    }

    // Start the sync thread, which will start the worker threads.
    SINFO("Launching sync thread '" << _syncThreadName << "'");
    _syncThread = thread(&BedrockServer::syncWrapper, this);
}

BedrockServer::~BedrockServer()
{
    // Shut down the sync thread, (which will shut down worker threads in turn).
    SINFO("Closing sync thread '" << _syncThreadName << "'");
    if (_syncThread.joinable()) {
        _syncThread.join();
    }
    SINFO("Threads closed.");

    if (_outstandingSocketThreads) {
        SWARN("Shutting down with " << _outstandingSocketThreads << " socket threads remaining.");
    }

    // Delete our plugins.
    for (auto& p : plugins) {
        delete p.second;
    }

    shutdownTimer.serverDestructor = chrono::steady_clock::now();
    SINFO("Shutdown timing: "
          << "start->safeState=" << chrono::duration_cast<chrono::milliseconds>(shutdownTimer.safeNodeState - shutdownTimer.shutdownStart)
          << ", safeState->commandsComplete=" << chrono::duration_cast<chrono::milliseconds>(shutdownTimer.commandsComplete - shutdownTimer.shutdownStart)
          << ", commandsComplete->destructor=" << chrono::duration_cast<chrono::milliseconds>(shutdownTimer.serverDestructor - shutdownTimer.commandsComplete)
          << ".");
}

bool BedrockServer::shutdownComplete()
{
    if (_detach) {
        return shutdownWhileDetached;
    }

    // We're done when the sync thread is done.
    return !_syncLoopShouldBeRunning;
}

void BedrockServer::prePoll(fd_map& fdm)
{
    lock_guard<mutex> lock(_portMutex);

    // This will interrupt poll when we shut down.
    _notifyDone.prePoll(fdm);

    // Add all our ports. There are no sockets directly managed here.
    if (_commandPortPublic) {
        SFDset(fdm, _commandPortPublic->s, SREADEVTS);
    }
    if (_commandPortPrivate) {
        SFDset(fdm, _commandPortPrivate->s, SREADEVTS);
    }
    if (_controlPort) {
        SFDset(fdm, _controlPort->s, SREADEVTS);
    }
    for (const auto& p : _portPluginMap) {
        SFDset(fdm, p.first->s, SREADEVTS);
    }
}

void BedrockServer::postPoll(fd_map& fdm, uint64_t& nextActivity)
{
    _notifyDone.postPoll(fdm);

    // NOTE: There are no sockets managed here, just ports.
    // Open the port the first time we enter a command-processing state
    {
        lock_guard<mutex> lock(_portMutex);
        if (_commandPortBlockReasons.empty() && (getState() == SQLiteNodeState::LEADING || getState() == SQLiteNodeState::FOLLOWING) && _shutdownState.load() == RUNNING) {
            // Open the port
            if (!_commandPortPublic) {
                SINFO("Ready to process commands, opening public command port on '" << args["-serverHost"] << "'");
                _commandPortPublic = openPort(args["-serverHost"]);
            }
            if (!_commandPortPrivate) {
                SINFO("Ready to process commands, opening private command port on '" << args["-commandPortPrivate"] << "'");
                _commandPortPrivate = openPort(args["-commandPortPrivate"]);
            }
            if (!_controlPort) {
                SINFO("Opening control port on '" << args["-controlPort"] << "'");
                _controlPort = openPort(args["-controlPort"]);
            }

            // Open any plugin ports on enabled plugins
            for (auto plugin : plugins) {
                string portHost = plugin.second->getPort();
                if (!portHost.empty()) {
                    bool alreadyOpened = false;
                    for (auto& pluginPorts : _portPluginMap) {
                        if (pluginPorts.second == plugin.second) {
                            // We've already got this one.
                            alreadyOpened = true;
                            break;
                        }
                    }
                    // Open the port and associate it with the plugin
                    if (!alreadyOpened) {
                        SINFO("Opening port '" << portHost << "' for plugin '" << plugin.second->getName() << "'");
                        auto port = openPort(portHost);
                        if (port.get()) {
                            _portPluginMap[move(port)] = plugin.second;
                        } else {
                            SWARN("Failed opening port '" << portHost << "' for plugin '" << plugin.second->getName() << "'");
                        }
                    }
                }
            }
        }
    }

    // **NOTE: We leave the port open between startup and shutdown, even if we enter a state where
    //         we can't process commands -- such as a non leader/follower state.  The reason is we
    //         expect any state transitions between startup/shutdown to be due to temporary conditions
    //         that will resolve themselves automatically in a short time.  During this period we
    //         prefer to receive commands and queue them up, even if we can't process them immediately,
    //         on the assumption that we'll be able to process them before the browser times out.

    // Is the OS trying to communicate with us?
    if (SGetSignals()) {
        _beginShutdown(SGetSignalDescription());
    }

    // Accept any new connections
    _acceptSockets();

    // If any plugin timers are firing, let the plugins know.
    for (auto plugin : plugins) {
        for (SStopwatch* timer : plugin.second->timers) {
            if (timer->ding()) {
                plugin.second->timerFired(timer);
            }
        }
    }

    // If we've been told to start shutting down, we'll set the shut down timer.
    if (_shutdownState.load() == START_SHUTDOWN) {
        // Locking here means that no commands can be running when we do these checks and then switch to
        // `COMMANDS_FINISHED` because we have a shared lock on this mutex in `handleSocket`. This means this check can
        // only run between commands, and `_outstandingSocketThreads` will have been incremented already when we check
        // it here. So, if the check below for `_outstandingSocketThreads` passes at this point, it means there are no
        // commands at this point in time. However, new commands may still be received on the control port after this,
        // if we are detaching.
        unique_lock<shared_mutex> lock(_controlPortExclusionMutex);

        size_t count = BedrockCommand::getCommandCount();
        SINFO("SHUTDOWN Have " << _outstandingSocketThreads << " socket threads and " << count << " commands remaining.");

        // Don't tell the sync node to shut down while we still have commands or sockets left.
        if (!_outstandingSocketThreads && !count) {
            _shutdownState.store(COMMANDS_FINISHED);
            shutdownTimer.commandsComplete = chrono::steady_clock::now();

            // This interrupts the sync thread's poll() loop so it doesn't wait for up to an extra second to finish.
            // When it wakes up, it will begin its own shutdown.
            if (_syncNode) {
                _syncNode->notifyCommit();
            }
        }
    }
}

unique_ptr<BedrockCommand> BedrockServer::getCommandFromPlugins(SData&& request)
{
    auto baseCommand = make_unique<SQLiteCommand>(move(request));

    // Assign the command a cluster-unique id up front, so the plugin's command constructor can rely on command.id
    // being populated (it's moved into the constructed command before that command's constructor body runs). An
    // escalated command carries its originating id in the request's ID header, so we reuse that; otherwise we mint a
    // new one. The id is stored only on the command object, never written back into the request, so it can't leak into
    // request logging or into commands that build crash fingerprints from their request headers.
    const string commandID = baseCommand->request.isSet("ID") ? baseCommand->request["ID"] : generateCommandID();
    baseCommand->id = commandID;

    auto command = getCommandFromPlugins(move(baseCommand));

    // A plugin may return a freshly-built command rather than the base we passed in (e.g. an error command created
    // when a command's constructor throws), so make sure the returned command carries the id regardless.
    command->id = commandID;
    return command;
}

unique_ptr<BedrockCommand> BedrockServer::getCommandFromPlugins(unique_ptr<SQLiteCommand>&& baseCommand)
{
    try {
        for (auto pair : plugins) {
            // This is a bit weird to avoid changing this signature in all the plugins. It would be more straightforward if
            // the plugins just accepted a `unique_ptr<SQLiteCommand>&&`, but this still works.
            auto command = pair.second->getCommand(move(*baseCommand));
            if (command) {
                SDEBUG("Plugin " << pair.first << " handling command " << command->request.methodLine);
                return command;
            }
        }
    } catch (const SException& e) {
        SWARN("ENSURE_BUGBOT Command constructor failed with error: "s + e.what() + ", on command: " + baseCommand->request.methodLine);
        auto errorCommand = make_unique<BedrockCommand>(SQLiteCommand(), nullptr);
        errorCommand->complete = true;
        errorCommand->response.methodLine = e.what();
        if (!e.headers.empty()) {
            errorCommand->response.nameValueMap = e.headers;
        }
        if (!e.body.empty()) {
            errorCommand->response.content = e.body;
        }
        return errorCommand;
    } catch (const exception& e) {
        SWARN("ENSURE_BUGBOT Command constructor failed with error: "s + e.what() + ", on command: " + baseCommand->request.methodLine);
        auto errorCommand = make_unique<BedrockCommand>(SQLiteCommand(), nullptr);
        errorCommand->complete = true;
        errorCommand->response.methodLine = "500 Internal server error";
        return errorCommand;
    } catch (...) {
        SWARN("ENSURE_BUGBOT Command constructor failed with error: UNKNOWN, on command: " + baseCommand->request.methodLine);
        auto errorCommand = make_unique<BedrockCommand>(SQLiteCommand(), nullptr);
        errorCommand->complete = true;
        errorCommand->response.methodLine = "500 Internal server error";
        return errorCommand;
    }

    // Same weirdness as above, but for default commands.
    return make_unique<BedrockCommand>(move(*baseCommand), nullptr);
}

void BedrockServer::_reply(unique_ptr<BedrockCommand>& command)
{
    // Some control commands like Status go through a different flow where its SAUTOPREFIX scope has already closed, so we re-set it here so it handles those commands properly.
    SAUTOPREFIX(command->request);

    // Finalize timing info even for commands we won't respond to (this makes this data available in logs).
    command->finalizeTimingInfo();

    // Don't reply to commands with pseudo-clients (i.e., commands that we generated by other commands, or using
    // `Connection: forget`.
    if (command->initiatingClientID < 0) {
        command->handleFailedReply();
        return;
    }

    command->response["nodeName"] = args["-nodeName"];

    // If we're shutting down, tell the caller to close the connection.
    // Also, if the caller wanted us to close the connection, we'll parrot that back.
    if (_shutdownState.load() != RUNNING || command->request["Connection"] == "close") {
        command->response["Connection"] = "close";
    }

    // Is a plugin handling this command? If so, it gets to send the response.
    const string& pluginName = command->request["plugin"];

    if (command->socket) {
        if (!pluginName.empty()) {
            // Let the plugin handle it
            SINFO("Plugin '" << pluginName << "' handling response '" << command->response.methodLine
                  << "' to request '" << command->request.methodLine << "'");
            auto it = plugins.find(pluginName);
            if (it != plugins.end()) {
                it->second->onPortRequestComplete(*command, command->socket);
            } else {
                SERROR("Couldn't find plugin '" << pluginName << ".");
            }
        } else {
            // Otherwise we send the standard response.
            if (!command->socket->send(command->response.serialize())) {
                // If we can't send (client closed the socket?), alert our plugin it's response was never sent.
                SINFO("No socket to reply for: '" << command->request.methodLine << "' #" << command->initiatingClientID);
                command->handleFailedReply();
            }
        }

        // If `Connection: close` was set, shut down the socket, in case the caller ignores us.
        if (SIEquals(command->request["Connection"], "close") || _shutdownState.load() != RUNNING) {
            command->socket->shutdown();
        }
    } else {
        // This is the case for a fire-and-forget command, such as one set to run in the future. If `Connection:
        // forget` was specified, this is normal and we won't log.
        if (!SIEquals(command->request["Connection"], "forget")) {
            SINFO("No socket to reply for: '" << command->request.methodLine << "' #" << command->initiatingClientID);
        }
        command->handleFailedReply();
    }
}

void BedrockServer::blockCommandPort(const string& reason)
{
    lock_guard<mutex> lock(_portMutex);
    _commandPortBlockReasons.insert(reason);
    _isCommandPortLikelyBlocked = true;
    if (_commandPortBlockReasons.size() == 1) {
        _commandPortPublic = nullptr;
        _portPluginMap.clear();
    }
    SINFO("Blocking command port due to: " << reason << (_commandPortBlockReasons.size() > 1 ? " (already blocked)" : "") << ".");
}

void BedrockServer::unblockCommandPort(const string& reason)
{
    lock_guard<mutex> lock(_portMutex);
    auto it = _commandPortBlockReasons.find(reason);
    if (it == _commandPortBlockReasons.end()) {
        SWARN("Tried to remove command port block because: " << reason << ", but it wasn't blocked for that reason!");
    } else {
        _commandPortBlockReasons.erase(it);
        SINFO("Removing command port block due to: " << reason << (_commandPortBlockReasons.size() > 0 ? " (blocks remaining)" : "") << ".");
    }
    if (_commandPortBlockReasons.empty()) {
        _isCommandPortLikelyBlocked = false;
    }
}

bool BedrockServer::isCommandPortBlocked() const
{
    return _isCommandPortLikelyBlocked;
}

void BedrockServer::suppressCommandPort(const string& reason, bool suppress, bool manualOverride)
{
    if (suppress) {
        blockCommandPort("LEGACY_" + reason);
    } else {
        unblockCommandPort("LEGACY_" + reason);
    }
}

bool BedrockServer::_isStatusCommand(const unique_ptr<BedrockCommand>& command)
{
    if (SIEquals(command->request.methodLine, STATUS_IS_FOLLOWER) ||
        SIEquals(command->request.methodLine, STATUS_HANDLING_COMMANDS) ||
        SIEquals(command->request.methodLine, STATUS_PING) ||
        SIEquals(command->request.methodLine, STATUS_STATUS)) {
        return true;
    }
    return false;
}

list<STable> BedrockServer::getPeerInfo()
{
    list<STable> peerData;
    auto _syncNodeCopy = atomic_load(&_syncNode);
    if (_syncNodeCopy) {
        peerData = _syncNodeCopy->getPeerInfo();
    }
    return peerData;
}

const string& BedrockServer::getVersion() const
{
    return _version;
}

void BedrockServer::setDetach(bool detach)
{
    if (detach) {
        _beginShutdown("Detach", true);
    } else {
        _detach = false;
    }
}

bool BedrockServer::isDetached()
{
    return _detach && !_syncLoopShouldBeRunning && _pluginsDetached;
}

bool BedrockServer::dbReadyToHandleRequests()
{
    auto state = getState();
    if (state == SQLiteNodeState::FOLLOWING) {
        // If we are following someone else is in charge of upgrades, we can serve requests.
        return true;
    }
    if (state == SQLiteNodeState::LEADING || state == SQLiteNodeState::STANDINGDOWN) {
        // We are upgraded if we have ever completed an upgrade (including via no-op).
        // If we have not yet completed an upgrade (the first thing we do when first going LEADING), we need
        // to do that before we're ready to serve requests.
        return _upgradeCompleted;
    }

    // In any other state, we aren't ready.
    return false;
}

void BedrockServer::_status(unique_ptr<BedrockCommand>& command)
{
    const SData& request = command->request;
    SData& response = command->response;

    // We'll return whether or not this server is following.
    if (SIEquals(request.methodLine, STATUS_IS_FOLLOWER)) {
        // Used for liveness check for HAProxy. It's limited to HTTP style requests for it's liveness checks, so let's
        // pretend to be an HTTP server for this purpose. This allows us to load balance incoming requests.
        //
        // HAProxy interprets 2xx/3xx level responses as alive, 4xx/5xx level responses as dead.
        if (getState() == SQLiteNodeState::FOLLOWING) {
            response.methodLine = "HTTP/1.1 200 Following";
        } else {
            response.methodLine = "HTTP/1.1 500 Not Following. State=" + SQLiteNode::stateName(getState());
        }
    } else if (SIEquals(request.methodLine, STATUS_HANDLING_COMMANDS)) {
        // This is similar to the above check, and is used for letting HAProxy load-balance commands.

        const string leaderVersion = _leaderVersion.load();
        if (_shouldBackup) {
            response.methodLine = "HTTP/1.1 503 Backup In Progress";
        } else if (_detach) {
            response.methodLine = "HTTP/1.1 503 Detached";
        } else if (!leaderVersion.empty() && _version != leaderVersion) {
            // Only a genuine mismatch when the leader's version is known. While SYNCHRONIZING/SEARCHING the leader
            // version is empty, so fall through to the state-based check below to report the accurate state.
            response.methodLine = "HTTP/1.1 500 Mismatched version. Version=" + _version;
        } else {
            string method = "HTTP/1.1 ";

            if (getState() == SQLiteNodeState::FOLLOWING || getState() == SQLiteNodeState::LEADING || getState() == SQLiteNodeState::STANDINGDOWN) {
                method += "200";
            } else {
                method += "500";
            }
            response.methodLine = method + " " + SQLiteNode::stateName(getState());
        }
    }
    // All a ping message requires is some response.
    else if (SIEquals(request.methodLine, STATUS_PING)) {
        response.methodLine = "200 OK";
    }
    // This collects the current state of the server, which also includes some state from the underlying SQLiteNode.
    else if (SIEquals(request.methodLine, STATUS_STATUS)) {
        STable content;
        list<string> pluginList;
        for (auto plugin : plugins) {
            STable pluginData = plugin.second->getInfo();
            pluginData["name"] = plugin.second->getName();
            pluginList.push_back(SComposeJSONObject(pluginData));
        }
        content["isLeader"] = getState() == SQLiteNodeState::LEADING ? "true" : "false";
        content["plugins"] = SComposeJSONArray(pluginList);
        content["state"] = SQLiteNode::stateName(getState());
        content["version"] = _version;
        content["host"] = args["-nodeHost"];
        content["commandCount"] = BedrockCommand::getCommandCount();
        content["isDetached"] = isDetached() ? "true" : "false";

        {
            // Make it known if anything is known to cause crashes.
            shared_lock<decltype(_crashCommandMutex)> lock(_crashCommandMutex);
            vector<string> crashCommandListArray;

            size_t totalCount = 0;
            for (const auto& s : _crashCommands) {
                totalCount += s.second.size();

                vector<string> paramsArray;
                for (const STable& params : s.second) {
                    if (!params.empty()) {
                        paramsArray.push_back(SComposeJSONObject(params));
                    }
                }

                STable commandObject;
                commandObject[s.first] = SComposeJSONArray(paramsArray);
                crashCommandListArray.push_back(SComposeJSONObject(commandObject));
            }
            content["crashCommands"] = totalCount;
            content["crashCommandList"] = SComposeJSONArray(crashCommandListArray);
        }

        for (auto& p : _blockingCommandQueue.getState()) {
            content[p.first] = p.second;
        }

        // Coalesce all of the peer data into one value to return or return
        // an error message if we timed out getting the peerList data.
        list<string> peerList;
        list<STable> peerData = getPeerInfo();
        for (const STable& peerTable : peerData) {
            peerList.push_back(SComposeJSONObject(peerTable));
        }

        {
            lock_guard<mutex> lock(_portMutex);
            content["commandPortBlockReasons"] = SComposeJSONArray(_commandPortBlockReasons);
        }
        content["maxOutstandingWALFrames"] = to_string(_maxOutstandingWALFrames.load());

        content["peerList"] = SComposeJSONArray(peerList);
        content["queuedCommandList"] = SComposeJSONArray(_commandQueue.getRequestMethodLines());

        auto _syncNodeCopy = atomic_load(&_syncNode);
        if (_syncNodeCopy) {
            content["syncNodeAvailable"] = "true";
            // Set some information about this node.
            content["CommitCount"] = to_string(_syncNodeCopy->getCommitCount());
            content["priority"] = to_string(_syncNodeCopy->getPriority());
            content["outstandingFramesToCheckpoint"] = to_string(_syncNodeCopy->getOutstandingFramesToCheckpoint());
            _syncNodeCopy = nullptr;
        } else {
            content["syncNodeAvailable"] = "false";
        }

        auto dbPoolCopy = getDBPool();
        if (dbPoolCopy) {
            SQLiteScopedHandle dbScope(*dbPoolCopy, dbPoolCopy->getIndex());
            SQLite& db = dbScope.db();
            SQResult freelistResult, pageCountResult;
            db.read("PRAGMA freelist_count;", freelistResult);
            if (!freelistResult.empty()) {
                content["freelistCount"] = freelistResult[0][0];
            }
            db.read("PRAGMA page_count;", pageCountResult);
            if (!pageCountResult.empty()) {
                content["pageCount"] = pageCountResult[0][0];
            }
        }

        // Done, compose the response.
        response.methodLine = "200 OK";
        response.content = SComposeJSONObject(content);
    }
}

bool BedrockServer::_isControlCommand(const unique_ptr<BedrockCommand>& command)
{
    if (SIEquals(command->request.methodLine, "BeginBackup") ||
        SIEquals(command->request.methodLine, "SuppressCommandPort") ||
        SIEquals(command->request.methodLine, "ClearCommandPort") ||
        SIEquals(command->request.methodLine, "ClearCrashCommands") ||
        SIEquals(command->request.methodLine, "ConflictReport") ||
        SIEquals(command->request.methodLine, "Detach") ||
        SIEquals(command->request.methodLine, "Attach") ||
        SIEquals(command->request.methodLine, "SetConflictParams") ||
        SIEquals(command->request.methodLine, "SetConflictPageLocks") ||
        SIEquals(command->request.methodLine, "EnableSQLTracing") ||
        SIEquals(command->request.methodLine, "SetJournalDeleter") ||
        SIEquals(command->request.methodLine, "BlockWrites") ||
        SIEquals(command->request.methodLine, "UnblockWrites") ||
        SIEquals(command->request.methodLine, "SetMaxSocketThreads") ||
        SIEquals(command->request.methodLine, "SetMaxOutstandingWALFrames") ||
        SIEquals(command->request.methodLine, "SetBlockingQueueTimeRateLimit") ||
        SIEquals(command->request.methodLine, "ClearBlockingQueue") ||
        SIEquals(command->request.methodLine, "SetPriority") ||
        SIEquals(command->request.methodLine, "GetCommitHash") ||
        SIEquals(command->request.methodLine, "CRASH_COMMAND")
    ) {
        return true;
    }
    return false;
}

bool BedrockServer::_isNonSecureControlCommand(const unique_ptr<BedrockCommand>& command)
{
    // A list of non-secure control commands that can be run from another host
    // TODO: Have some other way to specify privileged commands that can be
    // sent from other nodes on the private command port.
    return SIEquals(command->request.methodLine, "SuppressCommandPort") ||
           SIEquals(command->request.methodLine, "ClearCommandPort") ||
           SIEquals(command->request.methodLine, "GetCommitHash") ||
           SIEquals(command->request.methodLine, "CRASH_COMMAND");
}

// State management for blocking writes to the DB.
mutex __quiesceLock;
atomic<bool> __quiesceShouldUnlock(false);
thread* __quiesceThread = nullptr;

void BedrockServer::_control(unique_ptr<BedrockCommand>& command)
{
    SINFO("Received control command: " << command->request.methodLine);
    SData& response = command->response;
    string reason = "MANUAL";
    response.methodLine = "200 OK";
    if (SIEquals(command->request.methodLine, "BeginBackup")) {
        _shouldBackup = true;
        _beginShutdown("Detach", true);
    } else if (SIEquals(command->request.methodLine, "SuppressCommandPort")) {
        if (command->request.isSet("reason") && command->request["reason"].size()) {
            reason = command->request["reason"];
        }
        blockCommandPort(reason);
    } else if (SIEquals(command->request.methodLine, "ClearCommandPort")) {
        if (command->request.isSet("reason") && command->request["reason"].size()) {
            reason = command->request["reason"];
        }
        unblockCommandPort(reason);
    } else if (SIEquals(command->request.methodLine, "ClearCrashCommands")) {
        unique_lock<decltype(_crashCommandMutex)> lock(_crashCommandMutex);
        _crashCommands.clear();
    } else if (SIEquals(command->request.methodLine, "ConflictReport")) {
        response.content = _conflictManager.generateReport();
    } else if (SIEquals(command->request.methodLine, "Detach")) {
        if (isDetached()) {
            response.methodLine = "400 Already detached";
        } else {
            response.methodLine = "203 DETACHING";
            _beginShutdown("Detach", true);
        }
    } else if (SIEquals(command->request.methodLine, "Attach")) {
        // Ensure none of our plugins are blocking attaching
        list<string> blockingPlugins;
        for (auto plugin : plugins) {
            if (plugin.second->preventAttach()) {
                blockingPlugins.emplace_back(plugin.second->getName());
            }
        }
        if (blockingPlugins.size()) {
            response.methodLine = "401 Attaching prevented by " + SComposeList(blockingPlugins);
        } else if (_shutdownState.load() != RUNNING) {
            // Wait to confirm that we're in the final _shutdownState "RUNNING" before reattaching
            response.methodLine = "401 Attaching prevented by server not ready";
        } else {
            response.methodLine = "204 ATTACHING";
            _syncLoopShouldBeRunning = true;
            _detach = false;
        }
    } else if (SIEquals(command->request.methodLine, "SetJournalDeleter")) {
        // Zero pauses trimming, but SQLite reads a negative limit as no limit at all, which would delete a whole
        // journal table in one transaction.
        const int64_t batchSize = command->request.calc64("batchSize");
        if (command->request.isSet("batchSize") && batchSize < 0) {
            response.methodLine = "400 batchSize can't be negative";
            return;
        }

        response["oldEnabled"] = BedrockJournalDeleter::enableDeleterThread ? "true" : "false";
        response["oldBatchSize"] = to_string(BedrockJournalDeleter::deleterBatchSize.load());
        if (command->request.isSet("batchSize")) {
            BedrockJournalDeleter::deleterBatchSize.store(batchSize);
        }
        if (command->request.isSet("enable")) {
            BedrockJournalDeleter::enableDeleterThread.store(command->request.test("enable"));

            // The thread otherwise only starts with the sync loop, and only stops on shutdown.
            if (BedrockJournalDeleter::enableDeleterThread) {
                _journalDeleter.start();
            } else {
                _journalDeleter.stop();
            }
        }
        response["newEnabled"] = BedrockJournalDeleter::enableDeleterThread ? "true" : "false";
        response["newBatchSize"] = to_string(BedrockJournalDeleter::deleterBatchSize.load());
    } else if (SIEquals(command->request.methodLine, "EnableSQLTracing")) {
        response["oldValue"] = SQLite::enableTrace ? "true" : "false";
        if (command->request.isSet("enable")) {
            SQLite::enableTrace.store(command->request.test("enable"));
            response["newValue"] = SQLite::enableTrace ? "true" : "false";
        }
    } else if (SIEquals(command->request.methodLine, "CRASH_COMMAND")) {
        _handleCrashCommand(command->request);
    } else if (SIEquals(command->request.methodLine, "ClearBlockingQueue")) {
        auto commands = _blockingCommandQueue.getAll();
        list<string> methodLines;
        for (const auto& cmd : commands) {
            methodLines.push_back(cmd->request.methodLine);
        }
        SINFO("ClearBlockingQueue: failing " << commands.size() << " commands: " << SComposeList(methodLines));
        for (auto& cmd : commands) {
            cmd->response.methodLine = "503 Service Unavailable";
            cmd->complete = true;
            _reply(cmd);
        }
        // `getAll` bypasses `_dequeue`, so reset identifier counts to match the now-empty queue.
        _blockingCommandQueue.clearRateLimits();
        response["flushedCount"] = to_string(commands.size());
    } else if (SIEquals(command->request.methodLine, "SetConflictParams")) {
        int64_t maxConflictRetries = command->request.calc64("MaxConflictRetries");
        if (maxConflictRetries >= 0) {
            SINFO("Setting _maxConflictRetries to " << _maxConflictRetries);
            response["previousMaxConflictRetries"] = to_string(_maxConflictRetries.load());
            _maxConflictRetries.store(maxConflictRetries);
        }
    } else if (SIEquals(command->request.methodLine, "SetConflictPageLocks")) {
        _enableConflictPageLocks = command->request.test("enable");
    } else if (SIEquals(command->request.methodLine, "SetBlockingQueueTimeRateLimit")) {
        if (command->request.isSet("windowMS")) {
            int64_t windowMS = command->request.calc64("windowMS");
            if (windowMS >= 0) {
                uint64_t previousUS = _blockingCommandQueue.setSharedRateLimiterWindow(static_cast<uint64_t>(windowMS) * 1000);
                response["previousBlockingQueueWindowMS"] = to_string(previousUS / 1000);
                SINFO("Setting blocking queue rate limit window to " << windowMS << "ms");
            }
        }
        if (command->request.isSet("identifierThresholdMS")) {
            int64_t thresholdMS = command->request.calc64("identifierThresholdMS");
            if (thresholdMS >= 0) {
                uint64_t previousUS = _blockingCommandQueue.setBlockingIdentifierThreshold(static_cast<uint64_t>(thresholdMS) * 1000);
                response["previousBlockingQueueIdentifierThresholdMS"] = to_string(previousUS / 1000);
                SINFO("Setting blocking queue identifier threshold to " << thresholdMS << "ms");
            }
        }
        if (command->request.isSet("commandThresholdMS")) {
            int64_t thresholdMS = command->request.calc64("commandThresholdMS");
            if (thresholdMS >= 0) {
                uint64_t previousUS = _blockingCommandQueue.setBlockingCommandThreshold(static_cast<uint64_t>(thresholdMS) * 1000);
                response["previousBlockingQueueCommandThresholdMS"] = to_string(previousUS / 1000);
                SINFO("Setting blocking queue command threshold to " << thresholdMS << "ms");
            }
        }
        if (command->request.isSet("blockDurationMS")) {
            int64_t durationMS = command->request.calc64("blockDurationMS");
            if (durationMS >= 0) {
                uint64_t previousUS = _blockingCommandQueue.setSharedRateLimiterBlockDuration(static_cast<uint64_t>(durationMS) * 1000);
                response["previousBlockingQueueBlockDurationMS"] = to_string(previousUS / 1000);
                SINFO("Setting blocking queue block duration to " << durationMS << "ms");
            }
        }
        if (command->request.isSet("globalRateLimiterWindowMS")) {
            int64_t windowMS = command->request.calc64("globalRateLimiterWindowMS");
            if (windowMS >= 0) {
                uint64_t previousUS = _blockingCommandQueue.setGlobalRateLimiterWindow(static_cast<uint64_t>(windowMS) * 1000);
                response["previousGlobalRateLimiterWindowMS"] = to_string(previousUS / 1000);
                SINFO("Setting blocking queue global rate limiter window to " << windowMS << "ms");
            }
        }
        if (command->request.isSet("globalRateLimiterThresholdMS")) {
            int64_t thresholdMS = command->request.calc64("globalRateLimiterThresholdMS");
            if (thresholdMS >= 0) {
                uint64_t previousUS = _blockingCommandQueue.setGlobalRateLimiterThreshold(static_cast<uint64_t>(thresholdMS) * 1000);
                response["previousGlobalRateLimiterThresholdMS"] = to_string(previousUS / 1000);
                SINFO("Setting blocking queue global rate limiter threshold to " << thresholdMS << "ms");
            }
        }
        if (command->request.isSet("globalRateLimiterBlockDurationMS")) {
            int64_t durationMS = command->request.calc64("globalRateLimiterBlockDurationMS");
            if (durationMS >= 0) {
                uint64_t previousUS = _blockingCommandQueue.setGlobalRateLimiterBlockDuration(static_cast<uint64_t>(durationMS) * 1000);
                response["previousGlobalRateLimiterBlockDurationMS"] = to_string(previousUS / 1000);
                SINFO("Setting blocking queue global rate limiter block duration to " << durationMS << "ms");
            }
        }
        if (command->request.test("ClearBlocks")) {
            _blockingCommandQueue.clearRateLimits();
        }
    } else if (SIEquals(command->request.methodLine, "BlockWrites")) {
        atomic<bool> locked(false);
        lock_guard lock(__quiesceLock);
        if (__quiesceThread) {
            response.methodLine = "400 Already Blocked";
        } else {
            __quiesceThread = new thread([&]() {
                shared_ptr<SQLitePool> dbPoolCopy = getDBPool();
                if (dbPoolCopy) {
                    SQLiteScopedHandle dbScope(*dbPoolCopy, dbPoolCopy->getIndex());
                    SQLite& db = dbScope.db();
                    SINFO("[quiesce] Exclusive locking DB");
                    db.exclusiveLockDB();
                    SINFO("[quiesce] Exclusive locked DB");
                    locked = true;
                    while (true) {
                        if (__quiesceShouldUnlock) {
                            db.exclusiveUnlockDB();
                            __quiesceShouldUnlock = false;
                            return;
                        }

                        // Wait 10ms for the next check.
                        usleep(10'000);
                    }
                }
            });

            // Repeatedly wait 10ms for the lock until the thread indicates it's been acquired.
            while (locked == false) {
                usleep(10'000);
            }

            response.methodLine = "200 Blocked";
        }
    } else if (SIEquals(command->request.methodLine, "UnblockWrites")) {
        SINFO("[quiesce] Locking __quiesceLock");
        lock_guard lock(__quiesceLock);
        SINFO("[quiesce] __quiesceLock locked");
        if (!__quiesceThread) {
            response.methodLine = "200 Not Blocked";
        } else {
            __quiesceShouldUnlock = true;
            SINFO("[quiesce] Joining __quiesceThread");
            __quiesceThread->join();
            SINFO("[quiesce] __quiesceThread joined");
            delete __quiesceThread;
            __quiesceThread = nullptr;
            response.methodLine = "200 Unblocked";
        }
    } else if (SIEquals(command->request.methodLine, "SetMaxSocketThreads")) {
        size_t newMax = command->request.calcU64("socketThreadsCount");
        if (newMax) {
            SINFO("Setting _maxSocketThreads to " << newMax << " from " << _maxSocketThreads);
            _maxSocketThreads = newMax;
        } else {
            response.methodLine = "401 Don't Use Zero";
        }
    } else if (SIEquals(command->request.methodLine, "SetMaxOutstandingWALFrames")) {
        const int64_t newMax = command->request.calc64("maxOutstandingWALFrames");
        if (!command->request.isSet("maxOutstandingWALFrames")) {
            response.methodLine = "400 Missing maxOutstandingWALFrames";
        } else if (newMax < 0) {
            response.methodLine = "400 maxOutstandingWALFrames can't be negative";
        } else {
            const uint64_t oldMax = _maxOutstandingWALFrames.exchange(static_cast<uint64_t>(newMax));
            SINFO("Setting _maxOutstandingWALFrames to " << newMax << " from " << oldMax);
            response["oldMaxOutstandingWALFrames"] = to_string(oldMax);
            response["newMaxOutstandingWALFrames"] = to_string(newMax);
        }
    } else if (SIEquals(command->request.methodLine, "SetPriority")) {
        if (!command->request.isSet("priority")) {
            response.methodLine = "400 Missing priority";
        } else {
            int64_t newPriority = command->request.calc64("priority");
            auto _syncNodeCopy = atomic_load(&_syncNode);
            // Priority 1 is reserved for shutdown.
            if (newPriority < 0 || newPriority == 1) {
                response.methodLine = "400 Invalid priority";
            } else if (!_syncNodeCopy) {
                response.methodLine = "500 No sync node";
            } else {
                try {
                    int oldPriority = _syncNodeCopy->setPriority(newPriority);
                    response["oldPriority"] = to_string(oldPriority);
                    response["newPriority"] = to_string(newPriority);
                    response.methodLine = "200 OK";
                } catch (const SException& e) {
                    response.methodLine = e.what();
                }
            }
        }
    } else if (SIEquals(command->request.methodLine, "GetCommitHash")) {
        if (!command->request.isSet("commitCount")) {
            response.methodLine = "400 Missing commitCount";
        } else {
            // Journal IDs start at 1, so 0 (which is also what we get for anything unparseable) is never valid.
            uint64_t commitCount = command->request.calcU64("commitCount");
            shared_ptr<SQLitePool> dbPoolCopy = getDBPool();
            if (!commitCount) {
                response.methodLine = "400 Invalid commitCount";
            } else if (!dbPoolCopy) {
                response.methodLine = "500 No database";
            } else {
                SQLiteScopedHandle dbScope(*dbPoolCopy, dbPoolCopy->getIndex());
                string hash;
                if (dbScope.db().getCommit(commitCount, nullptr, &hash)) {
                    response["commitCount"] = to_string(commitCount);
                    response["hash"] = hash;
                } else {
                    response.methodLine = "404 Commit not found";
                }
            }
        }
    }
}

SData BedrockServer::BedrockServerUpgradeCommand::_buildRequest()
{
    SData request("UpgradeDB");

    // Nobody is waiting on a response to this, the sync thread learns that it finished from `destructionCallback`.
    request["Connection"] = "forget";

    return request;
}

BedrockServer::BedrockServerUpgradeCommand::BedrockServerUpgradeCommand(BedrockServer& server)
    : BedrockCommand(SQLiteCommand(_buildRequest()), nullptr), _server(server)
{
    // Schema changes run in an exclusive transaction, which otherwise gets the five second budget that any command
    // not named in `-synchronousCommands` gets. Take the twenty seconds those commands get instead.
    isSynchronous = true;
}

bool BedrockServer::BedrockServerUpgradeCommand::peek(SQLite& db)
{
    return false;
}

void BedrockServer::BedrockServerUpgradeCommand::process(SQLite& db)
{
    // These all get conglomerated into one big query.
    try {
        for (auto plugin : _server.plugins) {
            plugin.second->upgradeDatabase(db);
        }
    } catch (const system_error& e) {
        SWARN("Caught system_error upgrading DB, code: " << e.code() << ", message: " << e.what());
        throw;
    }
    SINFO("Finished running DB upgrade.");
}

void BedrockServer::_beginShutdown(const string& reason, bool detach)
{
    if (_shutdownState.load() == RUNNING) {
        _detach = detach;
        // Begin a graceful shutdown; close our port
        SINFO("Beginning graceful shutdown due to '" << reason << "', closing command port on '" << args["-serverHost"] << "'.");
        if (args.isSet("-testShutdownReasonFile")) {
            SFileSave(args["-testShutdownReasonFile"], reason);
        }
        shutdownTimer.shutdownStart = chrono::steady_clock::now();
        if (getState() != SQLiteNodeState::LEADING && getState() != SQLiteNodeState::STANDINGDOWN) {
            shutdownTimer.safeNodeState = shutdownTimer.shutdownStart;
        }

        // Delete any commands scheduled in the future.
        _commandQueue.abandonFutureCommands(5000);

        // Accept any new connections before closing, this avoids leaving clients who had connected to in a weird
        // state.
        _acceptSockets();

        // Close our listening ports, we won't accept any new connections on them, except the control port, if we're
        // detaching. It needs to keep listening.
        // We lock around changing the shutdown state because `postPoll` will open these ports if we're not shutting
        // down, so otherwise there's a race condition where that happens just after we close them but before we
        // change the state.
        {
            lock_guard<mutex> lock(_portMutex);
            _commandPortPublic = nullptr;
            _commandPortPrivate = nullptr;
            if (!_detach) {
                _controlPort = nullptr;
            }
            _portPluginMap.clear();
            _shutdownState.store(START_SHUTDOWN);
        }
        SQLiteNodeState currentState = SQLiteNodeState::UNKNOWN;
        auto syncNodeCopy = atomic_load(&_syncNode);
        if (syncNodeCopy) {
            currentState = syncNodeCopy->getState();
            syncNodeCopy->setShutdownPriority();
        }

        // Set a 10 second DB timeout for all remaining queries.
        _shutdownTime.store(chrono::steady_clock::now() + chrono::seconds{10});
        SINFO("START_SHUTDOWN. Ports shutdown, will perform final socket read. Queries will time out in 10s. Commands queued: " << _commandQueue.size()
              << ", blocking commands queued: " << _blockingCommandQueue.size() << ", total commands: " << BedrockCommand::getCommandCount()
              << ", state: " << SQLiteNode::stateName(currentState));
    }
}

bool BedrockServer::shouldBackup()
{
    return _shouldBackup;
}

void BedrockServer::_handleCrashCommand(const SData& message)
{
    SData request;
    request.deserialize(message.content);
    if (request.empty()) {
        SINFO("Got CRASH_COMMAND with malformed command body. Is someone running this by hand? Nothing to blacklist.");
        return;
    }

    // Take a unique lock so nobody else can read from this table while we update it.
    unique_lock<decltype(_crashCommandMutex)> lock(_crashCommandMutex);

    // Strip Content-Length before storing to prevent the same identifiers from being stored twice.
    request.nameValueMap.erase("Content-Length");

    _crashCommands[request.methodLine].insert(request.nameValueMap);
    size_t totalCount = 0;
    for (const auto& s : _crashCommands) {
        totalCount += s.second.size();
    }
    SALERT("Blacklisting command (now have " << totalCount << " blacklisted commands): " << request.serialize());
}

SData BedrockServer::_generateCrashMessage(const unique_ptr<BedrockCommand>& command)
{
    SHMMM("Generating CRASH_COMMAND command for " << command->request.methodLine);
    SData message("CRASH_COMMAND");
    SData subMessage(command->request.methodLine);
    for (auto& pair : command->crashIdentifyingValues) {
        subMessage.emplace(pair);
    }
    message["timeout"] = "5000";
    message.content = subMessage.serialize();
    return message;
}

void BedrockServer::broadcastCommand(const SData& command, const string& peerName)
{
    auto _clusterMessengerCopy = atomic_load(&_clusterMessenger);
    if (!_clusterMessengerCopy) {
        SINFO("Failed to broadcast command " << command.methodLine << " to all nodes, cluster messenger does not exist.");
        return;
    }

    if (peerName == "") {
        _clusterMessengerCopy->runOnAll(command);
    } else {
        BedrockCommand cmd(SQLiteCommand(SData(command)), nullptr);
        bool result = _clusterMessengerCopy->runOnPeer(cmd, peerName);
        if (!result) {
            SWARN("Failed to run command " << command.methodLine << " on peer " << peerName);
        }
    }
    SINFO("Completed broadcast of command " << command.methodLine << ".");
}

void BedrockServer::onNodeLogin(SQLitePeer* peer)
{
    list<unique_ptr<BedrockCommand>> crashCommandsToSend;
    {
        shared_lock<decltype(_crashCommandMutex)> lock(_crashCommandMutex);
        for (const auto& p : _crashCommands) {
            for (const auto& table : p.second) {
                SALERT("Sending crash command " << p.first << " to node " << peer->name << " on login");
                SData crashCommandSpec(p.first);
                crashCommandSpec.nameValueMap = table;
                unique_ptr<BedrockCommand> crashCommand = getCommandFromPlugins(move(crashCommandSpec));
                for (const auto& fields : table) {
                    crashCommand->crashIdentifyingValues.insert(fields.first);
                }
                crashCommandsToSend.push_back(make_unique<BedrockCommand>(_generateCrashMessage(crashCommand), nullptr));
            }
        }
    }

    for (auto& peerCommand : crashCommandsToSend) {
        // The cluster connection is already established, even if the peer's private command port is still closed.
        _syncNode->sendMessageToPeer(peer, peerCommand->request);

        // Keep HTTP replay for older nodes that ignore cluster CRASH_COMMAND messages. Remove once all nodes support it.
        auto _clusterMessengerCopy = atomic_load(&_clusterMessenger);
        auto peerName = peer->name;
        if (_clusterMessengerCopy) {
            thread([command = move(peerCommand), _clusterMessengerCopy, peerName]() {
                _clusterMessengerCopy->runOnPeer(*command, peerName);
            }).detach();
        }
    }

    // Let each plugin send the joining peer any of its own state that needs to replicate.
    for (auto& p : plugins) {
        p.second->onNodeLogin(peer);
    }
}

void BedrockServer::_acceptSockets()
{
    // Try block because we sometimes catch `system_error` from in here (likely from the thread code) and we're
    // trying to diagnose exactly what's happening.
    static uint64_t lastLogged = 0;
    try {
        // Make a list of ports to accept on.
        // We'll check the control port, command port, and any plugin ports for new connections.
        list<reference_wrapper<const unique_ptr<Port>>> portList = {_controlPort, _commandPortPrivate, _commandPortPublic};

        // Lock _portMutex so suppressing the port does not cause it to be null
        // in the middle of this function.
        lock_guard<mutex> lock(_portMutex);

        for (auto& p : _portPluginMap) {
            portList.push_back(reference_wrapper<const unique_ptr<Port>>(p.first));
        }

        // Try each port.
        for (auto portWrapper : portList) {
            const unique_ptr<Port>& port = portWrapper.get();

            // Skip null ports (if the command or control port are closed).
            if (!port) {
                continue;
            }

            // Accept as many sockets as we can.
            while (true) {
                uint64_t now = STimeNow();
                if ((port != _controlPort) && (_outstandingSocketThreads >= _maxSocketThreads)) {
                    if ((lastLogged < now - 3'000'000)) {
                        SWARN("Not accepting any new socket threads as we already have " << _outstandingSocketThreads << " of " << _maxSocketThreads);
                        lastLogged = now;
                    }
                    return;
                }

                sockaddr_in addr;
                int s = S_accept(port->s, addr, true); // Note that this sets the newly accepted socket to be blocking.

                // If we got an error or no socket, done accepting for now.
                if (s <= 0) {
                    break;
                }

                // Otherwise create the object for this new socket.
                SDEBUG("Accepting socket from '" << addr << "' on port '" << port->host << "'");
                Socket socket(s, Socket::CONNECTED);
                socket.addr = addr;

                // If it came from a plugin, record that.
                auto plugin = _portPluginMap.find(port);
                if (plugin != _portPluginMap.end()) {
                    socket.data = plugin->second;
                    // Call the plugin's handler.
                    plugin->second->onPortAccept(&socket);
                }

                // And start up this socket's thread.
                _outstandingSocketThreads++;
                thread t;
                bool threadStarted = false;
                while (!threadStarted) {
                    try {
                        t = thread(&BedrockServer::handleSocket, this, move(socket), port == _controlPort, port == _commandPortPublic, port == _commandPortPrivate);
                        threadStarted = true;
                    } catch (const system_error& e) {
                        // We don't care about this lock here from a performance perspective, it only happens when we
                        // are unable to do any work anyway (i.e., we can't start threads).
                        lock_guard<mutex> lock(_newSocketThreadBlockedMutex);
                        if (_outstandingSocketThreads < 100) {
                            // We don't expect this to ever happen - we only seem to get `system_error` here when we
                            // have lots (thousands) of threads running. Because of this, our handling of this in
                            // `handleSocket` only works correctly if this happens with greater than 50 threads, and if
                            // we were to block new threads with less than 50 threads already running, we'd never
                            // unblock new threads. Instead, if that happens, we throw this error and crash (which was
                            // the behavior we saw here before handling `system_error`).
                            // We check for 100 threads here instead of the 50 we check for in `handleSocket` to
                            // minimize the risk of race conditions pushing this number through `50` (either up or
                            // down) as we're checking this. For such a race condition to happen here, we'd need to
                            // increment/decrement all the way from 50-100 (or vice versa) to hit such a race condition,
                            // which is theoretically possible but exceedingly unlikely.
                            SERROR("Got system_error creating thread with only " << _outstandingSocketThreads << " threads!");
                        }
                        if (!_shouldBlockNewSocketThreads) {
                            // Block any new socket threads and warn.
                            _shouldBlockNewSocketThreads = true;
                            SWARN("Caught system_error in thread constructor (with " << _outstandingSocketThreads
                                  << " threads): " << e.code() << ", message: " << e.what() << ", blocking new socket threads.");
                            blockCommandPort("NOT_ENOUGH_THREADS");
                        }

                        // We just loop until socket threads are unblocked.
                        SINFO("Waiting 1 more second for socket threads to be available.");
                        sleep(1);
                    }
                }
                try {
                    t.detach();
                } catch (const system_error& e) {
                    SALERT("Caught system_error in thread detach: " << e.code() << ", message: " << e.what());
                    throw;
                }
            }
        }
    } catch (const system_error& e) {
        SALERT("Caught system_error outside thread startup: " << e.code() << ", message: " << e.what());
        throw;
    }
}

string BedrockServer::generateCommandID()
{
    return args["-nodeName"] + "#" + to_string(_requestCount++);
}

unique_ptr<BedrockCommand> BedrockServer::buildCommandFromRequest(SData&& request, Socket& socket, bool shouldTreatAsLocalhost)
{
    SAUTOPREFIX(request);

    bool fireAndForget = false;
    if (SIEquals(request["Connection"], "forget") || (uint64_t) request.calc64("commandExecuteTime") > STimeNow()) {
        // Respond immediately to make it clear we successfully queued it. We won't store the socket on the command.
        SINFO("Firing and forgetting '" << request.methodLine << "'");
        SData response("202 Successfully queued");
        response["Connection"] = "close";
        socket.send(response.serialize());
        socket.shutdown(Socket::CLOSED);
        fireAndForget = true;

        // If we're shutting down, discard this command, we won't wait for the future.
        if (_shutdownState.load() != RUNNING) {
            SINFO("Not queuing future command '" << request.methodLine << "' while shutting down.");
            return nullptr;
        }
    }

    // Get the source ip of the command.
    char* ip = inet_ntoa(socket.addr.sin_addr);
    if (!shouldTreatAsLocalhost && ip != "127.0.0.1"s) {
        // Auth checks to see that this value is missing/blank as a security check, so we leave it out for anything
        // originating on 127.0.0.1, or for which we've specified shouldTreatAsLocalhost (which are commands escalated
        // via the private command port). This is the *only* use of the `_source` attribute, so the only consideration
        // we need to make for this.
        request["_source"] = ip;
    }

    // Pull any serialized https requests off the requests object to apply to the command.
    string serializedHTTPSRequests = request["httpsRequests"];
    string serializedData = request["serializedData"];
    request.erase("httpsRequests");
    request.erase("serializedData");

    // Create a command.
    unique_ptr<BedrockCommand> command = getCommandFromPlugins(move(request));

    // And we and keep track of the client that initiated this command, so we can respond later, except
    // if we received connection:forget in which case we don't respond later
    command->initiatingClientID = SIEquals(command->request["Connection"], "forget") ? -1 : socket.id;

    try {
        // Apply HTTPS requests.
        if (serializedHTTPSRequests.size()) {
            command->deserializeHTTPSRequests(serializedHTTPSRequests);
            SINFO("Deserialized " << command->httpsRequests.size() << " HTTPS requests for command " << command->request.methodLine << ".");
        }
        if (serializedData.size()) {
            command->deserializeData(serializedData);
        }
    } catch (const SException& e) {
        command->complete = true;
        command->response.methodLine = e.method;
    } catch (const exception& e) {
        command->complete = true;
        command->response.methodLine = e.what();
    } catch (...) {
        command->complete = true;
        command->response.methodLine = "500 UNKNOWN ERROR";
    }

    SDEBUG("Deserialized command " << command->request.methodLine);
    command->socket = fireAndForget ? nullptr : &socket;

    if (_synchronousCommands.find(command->request.methodLine) != _synchronousCommands.end()) {
        command->isSynchronous = true;
        SINFO("Running command " << command->request.methodLine << " on the blocking commit thread.");
    }

    return command;
}

void BedrockServer::handleSocket(Socket&& socket, bool fromControlPort, bool fromPublicCommandPort, bool fromPrivateCommandPort)
{
    try {
        shared_lock<shared_mutex> controlPortLock(_controlPortExclusionMutex, defer_lock);
        if (fromControlPort) {
            controlPortLock.lock();
        }

        // Initialize and get a unique thread ID.
        string threadName = "socket" + to_string(_socketThreadNumber++);
        SInitialize(threadName);
        SINFO("[performance] Socket thread starting");

        // This outer loop just runs until the entire socket life cycle is done, meaning it deserializes a command,
        // waits for it to get processed, deserializes another, etc, until the socket gets closed.
        // This whole block is largely duplicated from `postPoll` and modified to work on a single non-blocking socket.
        while (socket.state != STCPManager::Socket::CLOSED) {
            // We are going to call `poll` in a loop with only this one socket as a file descriptor.
            // The reason for this is because it's possible that a client is connected to us, and not sending us any data.
            // It may be waiting for it's own data before it can send us a request, or it may have just forgotten to
            // disconnect. In the normal case, this is no big deal, we can wait inside `recv` until it either sends us some
            // data or it disconnects. The exception is if we want to shut down. In that case, we need to know to close the
            // socket at some point, so what we do is `poll` with a 1 second timeout, and if we ever hit the timeout and
            // are in a `shutting down` state, then we finish up and exit. In any other case, we just wait in `poll` again
            // until we get some data or a disconnection.
            int pollResult = 0;
            struct pollfd pollStruct = {socket.s, POLLIN, 0};

            // As long as `poll` returns 0 we've timed out, indicating that we're still waiting for something to happen. In
            // that case, we'll loop again *unless* we're shutting down.
            while (!(pollResult = poll(&pollStruct, 1, 1'000))) {
                if (_shutdownState != RUNNING) {
                    SINFO("Socket thread exiting because no data and shutting down.");
                    socket.shutdown(Socket::CLOSED);
                    break;
                }
            }

            // If the above loop didn't close the socket due to inactivity at shutdown, let's handle the activity.
            if (socket.state != STCPManager::Socket::CLOSED) {
                if (pollResult < 0) {
                    // This is an exceptional case, we'll just kill the socket if this happens and let the client reconnect.
                    SINFO("Poll failed: " << strerror(errno));
                    socket.shutdown(Socket::CLOSED);
                } else {
                    // We've either got new data, or an error on the socket. Let's determine which by trying to read.
                    if (!socket.recv()) {
                        // If reading failed, then the socket was closed.
                        socket.shutdown(Socket::CLOSED);
                    }
                }
            }

            // Now, if the socket hasn't been closed, we'll try to handle the new data on it appropriately.
            if (socket.state == STCPManager::Socket::CONNECTED) {
                // If there's a request, we'll dequeue it.
                SData request;

                // If the socket is owned by a plugin, we let the plugin populate our request.
                BedrockPlugin* plugin = static_cast<BedrockPlugin*>(socket.data);
                if (plugin) {
                    // Call the plugin's handler.
                    plugin->onPortRecv(&socket, request);
                    if (!request.empty()) {
                        // If it populated our request, then we'll save the plugin name so we can handle the response.
                        request["plugin"] = plugin->getName();
                    }
                } else {
                    // Otherwise, handle any default request.
                    int requestSize = 0;
                    if (socket.recvBuffer.startsWithHTTPRequest()) {
                        requestSize = request.deserialize(socket.recvBuffer);
                        socket.recvBuffer.consumeFront(requestSize);
                    }

                    // If this socket was accepted from the public command port, and that's supposed to be closed now, set
                    // `Connection: close` so that we don't keep doing a bunch of activity on it.
                    if (requestSize && fromPublicCommandPort && _isCommandPortLikelyBlocked) {
                        request["Connection"] = "close";
                    }
                }

                // If we have a populated request, from either a plugin or our default handling, we'll queue up the
                // command.
                if (!request.empty()) {
                    // Make a command from our request.
                    unique_ptr<BedrockCommand> command = buildCommandFromRequest(move(request), socket, fromPrivateCommandPort);

                    if (!command) {
                        // If we couldn't build a command, this was some sort of unusual exception case (like trying to
                        // schedule a command in the future while shutting down). We can just give up.
                        SINFO("No command from request, closing socket.");
                        socket.shutdown(Socket::CLOSED);
                    } else if (!_handleIfStatusOrControlCommand(command)) {
                        if (fromControlPort && _shutdownState != RUNNING) {
                            // Don't handle non-control commands on the control port if we're shutting down. As the control
                            // port can remain open through shutdown (in the case of detaching) and can expect DB access,
                            // which is being turned off, these could cause weird crashes. Instead, just return an error.
                            command->response.methodLine = "500 Server Shutting Down";
                            _reply(command);
                        } else {
                            // If it's not handled by `_handleIfStatusOrControlCommand` we fall into the queuing logic.
                            // If the command has a socket (it's this socket) then we need to wait for it to finish before
                            // we can dequeue the next command, so that the responses all end up delivered in order.
                            // If a command *doesn't* have a socket, then that's a special case for a `fire and forget`
                            // command that was already responded to in `buildCommandFromRequest` and we can move on to the
                            // next thing immediately.
                            mutex m;
                            condition_variable cv;
                            atomic<bool> finished = false;

                            function<void()> callback = [&m, &cv, &finished]() {
                                // Lock the mutex above (which will be locked by this thread while we're queuing), which waits
                                // for `handleSocket` to release it's lock (by calling `wait`), and then notify the waiting
                                // socket thread.
                                lock_guard lock(m);
                                finished = true;
                                cv.notify_all();
                            };

                            // Ok, none of above synchronization code gets called unless the command has a socket to respond on.
                            bool hasSocket = command->socket;
                            if (hasSocket) {
                                // Set the destructor callback for when the command finishes.
                                command->destructionCallback = &callback;
                            }

                            // Running the command in a separate thread allows this thread to poll the client socket and if it is
                            // disconnected, abort the command.
                            atomic<bool>& commandShouldAbortFlag = command->shouldAbort;
                            uint64_t commandThreadStartTime = STimeNow();
                            thread commandThread(
                                [&, commandThreadStartTime]() {
                                SInitialize(threadName + "_cmd");
                                command->recordTiming(BedrockCommand::COMMAND_THREAD, commandThreadStartTime);
                                runCommand(move(command));
                            });

                            // Now that the command is running, we wait for it to complete (if it has a socket, and hasn't finished by the time we get to this point).
                            // When this happens, destructionCallback fires, sets `finished` to true, and we can move on to the next request.
                            unique_lock<mutex> lock(m);
                            while (!finished && hasSocket) {
                                struct pollfd disconnectCheck = {socket.s, (POLLIN | POLLRDHUP), 0};
                                if (poll(&disconnectCheck, 1, 0) > 0) {
                                    if (!commandShouldAbortFlag && disconnectCheck.revents & (POLLHUP | POLLERR | POLLNVAL | POLLRDHUP)) {
                                        SINFO("Socket disconnected with command running, aborting.");
                                        commandShouldAbortFlag = true;
                                    }
                                }

                                // We also force an abort if we're shutting down.
                                auto shutdownTime = _shutdownTime.load();
                                if (!commandShouldAbortFlag && shutdownTime != chrono::time_point<chrono::steady_clock>{} && chrono::steady_clock::now() >= shutdownTime) {
                                    SINFO("Aborting command past shutdown timeout limit.");
                                    commandShouldAbortFlag = true;
                                }
                                cv.wait_for(lock, chrono::seconds(1));
                            }

                            commandThread.join();
                        }
                    }
                }
            } else if (socket.state == STCPManager::Socket::SHUTTINGDOWN || socket.state == STCPManager::Socket::CLOSED) {
                // Do nothing here except prevent the warning below from firing. This loop should exit on the next
                // iteration.
            } else {
                SWARN("Socket in unhandled state: " << socket.state);
            }
        }
    } catch (const exception& e) {
        SALERT("handleSocket got exception: " << e.what());
    } catch (...) {
        SALERT("handleSocket got unknown exception");
    }

    // At this point out socket is closed and we can clean up.
    // Note that we never return early, we always want to hit this code and decrement our counter and clean up our socket.
    _outstandingSocketThreads--;
    SINFO("[performance] Socket thread complete (" << _outstandingSocketThreads << " remaining).");

    // Check to see if we need to unblock creating new socket threads. We do this each time we cross having 50 active
    // threads. We are guaranteed to hit this as the thread count decrements to 0, as _shouldBlockNewSocketThreads is
    // atomic and every value must be hit as threads complete.
    // Note that if we were to start blocking the command port for NOT_ENOUGH_THREADS with less than 50 threads, we
    // will never hit this unlock case, but we only ever see this problem with thousands of threads, so we don't have
    // to try and handle that case, and don't need to lock this mutex on every thread's completion then.
    if (_outstandingSocketThreads == 50) {
        lock_guard<mutex> lock(_newSocketThreadBlockedMutex);
        if (_shouldBlockNewSocketThreads) {
            _shouldBlockNewSocketThreads = false;
            unblockCommandPort("NOT_ENOUGH_THREADS");
        }
    }
}

void BedrockServer::notifyStateChangeToPlugins(SQLite& db, SQLiteNodeState newState)
{
    for (auto plugin : plugins) {
        plugin.second->stateChanged(db, newState);
    }
}

SQLiteNodeState BedrockServer::getState() const
{
    auto _syncNodeCopy = atomic_load(&_syncNode);
    if (_syncNodeCopy) {
        return _syncNodeCopy->getState();
    }

    // For historical reasons, we return "SEARCHING" instead of "UNKNOWN" when the node is not available.
    // Scripts and tests, as well as the status command, expect this result.
    return SQLiteNodeState::SEARCHING;
}