CCBC-1611: handle 0x0d CONFIG_ONLY status code

Treat the CONFIG_ONLY status as a signal to refresh configuration.

The new or failed over nodes are set into config-only mode, where all
data operatios will be failed with code 0x0d. It is possible that the
SDK might be using stale configuration and send requests to the node,
that is not part of the cluster anymore, so to work around this, the
library will update the configuration and retry the operation.

Change-Id: I32eab64cd99e04681a9111bc66e1e597663c8de7
Reviewed-on: https://review.couchbase.org/c/libcouchbase/+/197126
Reviewed-by: Brett Lawson <brett19@gmail.com>
Tested-by: Build Bot <build@couchbase.com>

Author: avsej

include/memcached/protocol_binary.h

@@ -113,6 +113,9 @@ typedef enum {
     /** The requested resource is locked */
     PROTOCOL_BINARY_RESPONSE_LOCKED = 0x09,
 
+    /** The node cannot execute operation, because the bucket is in config-only mode */
+    PROTOCOL_BINARY_RESPONSE_CONFIG_ONLY = 0x0d,
+
     /** The authentication context is stale. You should reauthenticate*/
     PROTOCOL_BINARY_RESPONSE_AUTH_STALE = 0x1f,
     /** Authentication failure (invalid user/password combination,

src/bootstrap.cc

@@ -265,7 +265,7 @@ lcb_STATUS Bootstrap::bootstrap(unsigned options)
                 "Not requesting a config refresh because of throttling parameters. Next refresh possible in %" PRIu64
                 "ms or %u errors. "
                 "See LCB_CNTL_CONFDELAY_THRESH and LCB_CNTL_CONFERRTHRESH to modify the throttling settings",
-                LCB_NS2US(next_ts - now) / 1000, (unsigned)errthresh - errcounter);
+                LCB_NS2US(next_ts - now) / 1000, errthresh - errcounter);
             return LCB_SUCCESS;
         }
     }

src/bucketconfig/bc_cccp.cc

@@ -171,14 +171,15 @@ void CccpProvider::stop_current_request(bool is_clean)
  */
 lcb_STATUS CccpProvider::expect_config_with_version(const lcb_host_t *origin, config_version requested)
 {
-
     auto current = parent->get_current_version();
     auto previous_expected = expected_config_version;
-    if (current < requested && previous_expected < requested) {
+    if (expected_config_version < requested) {
+        expected_config_version = requested;
+    }
+    if (current < expected_config_version) {
         /*
          * The requested version is newer than we've seen so far (both current and previously requested)
          */
-        expected_config_version = requested;
         if (has_pending_request()) {
             /*
              * Only one configuration request could be in-flight, but the version is stored in expected_config_version
@@ -303,9 +304,21 @@ lcb_STATUS lcb::clconfig::cccp_update(Provider *provider, const char *host, cons
 
 lcb_STATUS lcb::clconfig::schedule_get_config(Provider *provider, const lcb_host_t *origin, config_version version)
 {
+    if (provider->type != CLCONFIG_CCCP) {
+        return LCB_ERR_INVALID_ARGUMENT;
+    }
     return static_cast<CccpProvider *>(provider)->expect_config_with_version(origin, version);
 }
 
+lcb_STATUS lcb::clconfig::schedule_get_config(Provider *provider)
+{
+    if (provider->type != CLCONFIG_CCCP) {
+        return LCB_ERR_INVALID_ARGUMENT;
+    }
+    return static_cast<CccpProvider *>(provider)->schedule_next_request(LCB_SUCCESS, /* can_rollover */ true,
+                                                                        /* skip_if_push_supported */ false);
+}
+
 lcb_STATUS CccpProvider::update(const char *host, const std::string &config_json)
 {
     if (config_json.empty()) {

src/bucketconfig/clconfig.h

@@ -733,6 +733,8 @@ void cccp_update(const void *cookie, lcb_STATUS err, const lcb_host_t *origin, c
 
 lcb_STATUS schedule_get_config(Provider *provider, const lcb_host_t *origin, config_version version);
 
+lcb_STATUS schedule_get_config(Provider *provider);
+
 /**
  * @brief record status of SELECT_BUCKET command
  * @param cookie_

src/mcserver/mcserver.cc

@@ -502,6 +502,29 @@ void Server::handle_clustermap_notification(const MemcachedResponse &request)
                                        lcb::clconfig::config_version{epoch, revision});
 }
 
+void Server::handle_config_only(const mc_PACKET *oldpkt)
+{
+    const auto *address = has_valid_host() ? &get_host() : nullptr;
+
+    lcb_log(LOGARGS_T(DEBUG),
+            LOGFMT "The bucket is configured in config-only mode on " LCB_HOST_FMT
+                   ", refresh configuration and retry operation",
+            LOGID_T(), LCB_HOST_ARG(this->settings, address));
+
+    auto *cccp = instance->confmon->get_provider(lcb::clconfig::CLCONFIG_CCCP);
+    if (cccp != nullptr && cccp->enabled) {
+        lcb::clconfig::schedule_get_config(cccp);
+    } else {
+        lcb_log(LOGARGS_T(DEBUG), LOGFMT "CCCP configuration provider is not enabled, using next available provider",
+                LOGID_T());
+        instance->confmon->do_next_provider();
+    }
+
+    mc_PACKET *newpkt = mcreq_renew_packet(oldpkt);
+    newpkt->flags &= ~MCREQ_STATE_FLAGS;
+    instance->retryq->config_only_add((mc_EXPACKET *)newpkt);
+}
+
 /**
  * Handles requests that are initiated by the KV engine. It only happens when PROTOCOL_BINARY_FEATURE_DUPLEX (0x0c) is
  * enabled for the connection.
@@ -647,6 +670,12 @@ Server::ReadState Server::try_read(lcbio_CTX *ctx, rdb_IOROPE *ior)
         }
         DO_SWALLOW_PAYLOAD()
         goto GT_DONE;
+    } else if (status == PROTOCOL_BINARY_RESPONSE_CONFIG_ONLY) {
+        /* consume the header */
+        DO_ASSIGN_PAYLOAD()
+        handle_config_only(request);
+        DO_SWALLOW_PAYLOAD()
+        goto GT_DONE;
     } else if (status == PROTOCOL_BINARY_RESPONSE_UNKNOWN_COLLECTION ||
                status == PROTOCOL_BINARY_RESPONSE_UNKNOWN_SCOPE) {
         /* consume the header */

src/mcserver/mcserver.h

@@ -220,6 +220,7 @@ class Server : public mc_PIPELINE
     bool handle_unknown_collection(MemcachedResponse &resinfo, mc_PACKET *oldpkt);
     void handle_server_request(const MemcachedResponse &request);
     void handle_clustermap_notification(const MemcachedResponse &request);
+    void handle_config_only(const mc_PACKET *oldpkt);
 
     bool maybe_retry_packet(mc_PACKET *pkt, lcb_STATUS err, protocol_binary_response_status status);
     bool maybe_reconnect_on_fake_timeout(lcb_STATUS received_error);

src/retryq.cc

@@ -474,6 +474,11 @@ void RetryQueue::ucadd(mc_EXPACKET *pkt, lcb_STATUS orig_err, protocol_binary_re
     add(pkt, orig_err, status, nullptr, 0);
 }
 
+void RetryQueue::config_only_add(mc_EXPACKET *pkt)
+{
+    add(pkt, LCB_ERR_UNSUPPORTED_OPERATION, PROTOCOL_BINARY_RESPONSE_CONFIG_ONLY, nullptr, 0);
+}
+
 static void fallback_handler(mc_CMDQUEUE *cq, mc_PACKET *pkt)
 {
     auto *instance = reinterpret_cast<lcb_INSTANCE *>(cq->cqdata);

src/retryq.h

@@ -88,6 +88,7 @@ class RetryQueue
      */
     void nmvadd(mc_EXPACKET *detchpkt);
     void ucadd(mc_EXPACKET *pkt, lcb_STATUS orig_err, protocol_binary_response_status status);
+    void config_only_add(mc_EXPACKET *pkt);
 
     /**
      * @brief Retry all queued operations

src/vbucket/vbucket.c

@@ -934,7 +934,7 @@ char *lcbvb_save_json(lcbvb_CONFIG *cfg)
     cJSON *root = cJSON_CreateObject();
 
     switch (cfg->dtype) {
-        case  LCBVB_DIST_VBUCKET:
+        case LCBVB_DIST_VBUCKET:
             tmp = cJSON_CreateString("vbucket");
             break;
         case LCBVB_DIST_KETAMA: