1/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
2/*
3 *     Copyright 2010-2020 Couchbase, Inc.
4 *
5 *   Licensed under the Apache License, Version 2.0 (the "License");
6 *   you may not use this file except in compliance with the License.
7 *   You may obtain a copy of the License at
8 *
9 *       http://www.apache.org/licenses/LICENSE-2.0
10 *
11 *   Unless required by applicable law or agreed to in writing, software
12 *   distributed under the License is distributed on an "AS IS" BASIS,
13 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 *   See the License for the specific language governing permissions and
15 *   limitations under the License.
16 */
17#include "internal.h"
18#include "collections.h"
19#include "auth-priv.h"
20#include "connspec.h"
21#include "logging.h"
22#include "hostlist.h"
23#include "rnd.h"
24#include "http/http.h"
25#include "bucketconfig/clconfig.h"
26#include "metrics/caching_meter.hh"
27#ifdef LCB_USE_HDR_HISTOGRAM
28#include "metrics/logging_meter.hh"
29#endif
30#include <lcbio/iotable.h>
31#include <lcbio/ssl.h>
32#include "defer.h"
33
34#define LOGARGS(obj, lvl) (obj)->settings, "instance", LCB_LOG_##lvl, __FILE__, __LINE__
35
36using namespace lcb;
37
38LIBCOUCHBASE_API lcb_STATUS lcb_createopts_create(lcb_CREATEOPTS **options, lcb_INSTANCE_TYPE type)
39{
40    *options = (lcb_CREATEOPTS *)calloc(1, sizeof(lcb_CREATEOPTS));
41    (*options)->type = type;
42    return LCB_SUCCESS;
43}
44
45LIBCOUCHBASE_API lcb_STATUS lcb_createopts_destroy(lcb_CREATEOPTS *options)
46{
47    free(options);
48    return LCB_SUCCESS;
49}
50
51LIBCOUCHBASE_API lcb_STATUS lcb_createopts_connstr(lcb_CREATEOPTS *options, const char *connstr, size_t connstr_len)
52{
53    options->connstr = connstr;
54    options->connstr_len = connstr_len;
55    return LCB_SUCCESS;
56}
57
58LIBCOUCHBASE_API lcb_STATUS lcb_createopts_bucket(lcb_CREATEOPTS *options, const char *bucket, size_t bucket_len)
59{
60    options->bucket = bucket;
61    options->bucket_len = bucket_len;
62    return LCB_SUCCESS;
63}
64
65LIBCOUCHBASE_API lcb_STATUS lcb_createopts_logger(lcb_CREATEOPTS *options, const lcb_LOGGER *logger)
66{
67    options->logger = logger;
68    return LCB_SUCCESS;
69}
70
71LIBCOUCHBASE_API lcb_STATUS lcb_createopts_credentials(lcb_CREATEOPTS *options, const char *username,
72                                                       size_t username_len, const char *password, size_t password_len)
73{
74    options->username = username;
75    options->username_len = username_len;
76    options->password = password;
77    options->password_len = password_len;
78    return LCB_SUCCESS;
79}
80
81LIBCOUCHBASE_API lcb_STATUS lcb_createopts_authenticator(lcb_CREATEOPTS *options, lcb_AUTHENTICATOR *auth)
82{
83    options->auth = auth;
84    return LCB_SUCCESS;
85}
86
87LIBCOUCHBASE_API lcb_STATUS lcb_createopts_io(lcb_CREATEOPTS *options, struct lcb_io_opt_st *io)
88{
89    options->io = io;
90    return LCB_SUCCESS;
91}
92
93LIBCOUCHBASE_API lcb_STATUS lcb_createopts_tracer(lcb_CREATEOPTS *options, struct lcbtrace_TRACER *tracer)
94{
95    options->tracer = tracer;
96    return LCB_SUCCESS;
97}
98
99LIBCOUCHBASE_API lcb_STATUS lcb_createopts_meter(lcb_CREATEOPTS *options, const lcbmetrics_METER *meter)
100{
101    options->meter = meter;
102    return LCB_SUCCESS;
103}
104
105LIBCOUCHBASE_API
106const char *lcb_get_version(lcb_uint32_t *version)
107{
108    if (version != nullptr) {
109        *version = (lcb_uint32_t)LCB_VERSION;
110    }
111
112    return LCB_VERSION_STRING;
113}
114
115const lcb_U32 lcb_version_g = LCB_VERSION;
116
117LIBCOUCHBASE_API
118void lcb_set_cookie(lcb_INSTANCE *instance, const void *cookie)
119{
120    instance->cookie = cookie;
121}
122
123LIBCOUCHBASE_API
124const void *lcb_get_cookie(lcb_INSTANCE *instance)
125{
126    return instance->cookie;
127}
128
129LIBCOUCHBASE_API
130void lcb_set_auth(lcb_INSTANCE *instance, lcb_AUTHENTICATOR *auth)
131{
132    if (LCBT_SETTING(instance, keypath)) {
133        lcb_log(LOGARGS(instance, WARN),
134                "Custom authenticator ignored when SSL client certificate authentication in use");
135        return;
136    }
137    /* First increase refcount in case they are the same object(!) */
138    lcbauth_ref(auth);
139    lcbauth_unref(instance->settings->auth);
140    instance->settings->auth = auth;
141}
142
143void lcb_st::add_bs_host(const char *host, int port, unsigned bstype)
144{
145    const char *tname = nullptr;
146    lcb::Hostlist *target;
147    if (bstype == LCB_CONFIG_TRANSPORT_CCCP) {
148        tname = "CCCP";
149        target = mc_nodes;
150    } else {
151        tname = "HTTP";
152        target = ht_nodes;
153    }
154    bool ipv6 = strchr(host, ':') != nullptr;
155    lcb_log(LOGARGS(this, DEBUG), "Adding host " LCB_LOG_SPEC("%s%s%s:%d") " to initial %s bootstrap list",
156            this->settings->log_redaction ? LCB_LOG_SD_OTAG : "", ipv6 ? "[" : "", host, ipv6 ? "]" : "", port,
157            this->settings->log_redaction ? LCB_LOG_SD_CTAG : "", tname);
158    target->add(host, port);
159}
160
161void lcb_st::add_bs_host(const lcb::Spechost &host, int defl_http, int defl_cccp)
162{
163    if (host.isTypeless()) {
164        add_bs_host(host.hostname.c_str(), defl_http, LCB_CONFIG_TRANSPORT_HTTP);
165        add_bs_host(host.hostname.c_str(), defl_cccp, LCB_CONFIG_TRANSPORT_CCCP);
166        return;
167    } else {
168        add_bs_host(host.hostname.c_str(), host.port,
169                    host.isAnyHttp() ? LCB_CONFIG_TRANSPORT_HTTP : LCB_CONFIG_TRANSPORT_CCCP);
170    }
171}
172
173void lcb_st::populate_nodes(const Connspec &spec)
174{
175    int has_ssl = settings->sslopts & LCB_SSL_ENABLED;
176    int defl_http, defl_cccp;
177
178    if (spec.default_port() == LCB_CONFIG_MCCOMPAT_PORT) {
179        defl_http = -1;
180        defl_cccp = LCB_CONFIG_MCCOMPAT_PORT;
181
182    } else if (has_ssl) {
183        defl_http = LCB_CONFIG_HTTP_SSL_PORT;
184        defl_cccp = LCB_CONFIG_MCD_SSL_PORT;
185    } else {
186        defl_http = LCB_CONFIG_HTTP_PORT;
187        defl_cccp = LCB_CONFIG_MCD_PORT;
188    }
189
190    for (const auto &dh : spec.hosts()) {
191        add_bs_host(dh, defl_http, defl_cccp);
192    }
193    lcb_log(LOGARGS(this, TRACE), "Bootstrap hosts loaded (cccp:%d, http:%d)", (int)mc_nodes->size(),
194            (int)ht_nodes->size());
195}
196
197lcb_STATUS lcb_st::process_dns_srv(Connspec &spec)
198{
199    if (!spec.can_dnssrv()) {
200        return LCB_SUCCESS;
201    }
202    if (spec.hosts().empty()) {
203        lcb_log(LOGARGS(this, ERR), "Cannot use DNS SRV without a hostname");
204        return spec.is_explicit_dnssrv() ? LCB_ERR_INVALID_ARGUMENT : LCB_SUCCESS;
205    }
206
207    const Spechost &host = spec.hosts().front();
208    lcb_STATUS rc = LCB_ERR_SDK_INTERNAL;
209    Hostlist *hl = dnssrv_getbslist(host.hostname.c_str(), spec.sslopts() & LCB_SSL_ENABLED, rc);
210
211    if (hl == nullptr) {
212        lcb_log(LOGARGS(this, INFO), "DNS SRV lookup failed: %s. Ignore this if not relying on DNS SRV records",
213                lcb_strerror_short(rc));
214        if (spec.is_explicit_dnssrv()) {
215            return rc;
216        } else {
217            return LCB_SUCCESS;
218        }
219    }
220
221    spec.clear_hosts();
222    for (size_t ii = 0; ii < hl->size(); ++ii) {
223        const lcb_host_t &src = (*hl)[ii];
224        Spechost sh;
225        sh.hostname = src.host;
226        char *end = nullptr;
227        errno = 0;
228        long val = std::strtol(src.port, &end, 10);
229        if (errno == ERANGE || end == src.port) {
230            sh.port = 0;
231        } else {
232            sh.port = static_cast<std::uint16_t>(val);
233        }
234        sh.type = spec.default_port();
235        spec.add_host(sh);
236        bool ipv6 = sh.hostname.find(':') != std::string::npos;
237        lcb_log(LOGARGS(this, INFO), "Found host %s%s%s:%d via DNS SRV", ipv6 ? "[" : "", sh.hostname.c_str(),
238                ipv6 ? "]" : "", (int)sh.port);
239    }
240    delete hl;
241
242    return LCB_SUCCESS;
243}
244
245static lcb_STATUS init_providers(lcb_INSTANCE *obj, const Connspec &spec)
246{
247    using namespace lcb::clconfig;
248    Provider *http, *cccp, *mcraw;
249    http = obj->confmon->get_provider(CLCONFIG_HTTP);
250    cccp = obj->confmon->get_provider(CLCONFIG_CCCP);
251    mcraw = obj->confmon->get_provider(CLCONFIG_MCRAW);
252
253    if (spec.default_port() == LCB_CONFIG_MCCOMPAT_PORT) {
254        obj->confmon->set_active(CLCONFIG_MCRAW, true);
255        mcraw->configure_nodes(*obj->mc_nodes);
256        return LCB_SUCCESS;
257    }
258
259    bool cccp_found = spec.is_bs_cccp();
260    bool http_found = spec.is_bs_http();
261    bool cccp_enabled = true, http_enabled = true;
262
263    if (spec.is_bs_file()) {
264        cccp_found = false;
265        http_found = false;
266    }
267
268    if (cccp_found || http_found || spec.is_bs_file()) {
269        http_enabled = http_found;
270        cccp_enabled = cccp_found;
271    }
272
273    if (lcb_getenv_boolean("LCB_NO_CCCP")) {
274        cccp_enabled = false;
275    }
276    if (lcb_getenv_boolean("LCB_NO_HTTP")) {
277        http_enabled = false;
278    }
279
280    if (cccp_enabled == 0 && http_enabled == 0) {
281        if (spec.is_bs_file()) {
282            /* If the 'file_only' provider is set, just assume something else
283             * will provide us with the config, and forget about it. */
284            Provider *prov = obj->confmon->get_provider(CLCONFIG_FILE);
285            if (prov && prov->enabled) {
286                return LCB_SUCCESS;
287            }
288        }
289        if (obj->settings->conntype == LCB_TYPE_CLUSTER) {
290            /* Cluster-level connection always falls back to static config */
291            Provider *cladmin;
292            cladmin = obj->confmon->get_provider(CLCONFIG_CLADMIN);
293            cladmin->enable();
294            cladmin->configure_nodes(*obj->ht_nodes);
295        } else {
296            return LCB_ERR_BAD_ENVIRONMENT;
297        }
298    }
299
300    if (http_enabled) {
301        http->enable();
302        http->configure_nodes(*obj->ht_nodes);
303    } else {
304        obj->confmon->set_active(CLCONFIG_HTTP, false);
305    }
306
307    if (cccp_enabled) {
308        cccp->enable(obj);
309        cccp->configure_nodes(*obj->mc_nodes);
310    } else {
311        obj->confmon->set_active(CLCONFIG_CCCP, false);
312    }
313    return LCB_SUCCESS;
314}
315
316static lcb_STATUS setup_ssl(lcb_INSTANCE *obj, const Connspec &params)
317{
318    char optbuf[4096];
319    long env_policy = -1;
320    lcb_settings *settings = obj->settings;
321    lcb_STATUS err = LCB_SUCCESS;
322
323    if (lcb_getenv_nonempty("LCB_SSL_CACERT", optbuf, sizeof optbuf)) {
324        lcb_log(LOGARGS(obj, INFO), "SSL CA certificate %s specified on environment", optbuf);
325        settings->certpath = lcb_strdup(optbuf);
326    }
327
328    if (lcb_getenv_nonempty("LCB_SSL_KEY", optbuf, sizeof optbuf)) {
329        lcb_log(LOGARGS(obj, INFO), "SSL key %s specified on environment", optbuf);
330        settings->keypath = lcb_strdup(optbuf);
331    }
332
333    if (lcb_getenv_nonempty("LCB_SSL_MODE", optbuf, sizeof optbuf)) {
334        char *end = nullptr;
335        errno = 0;
336        env_policy = std::strtol(optbuf, &end, 10);
337        if (errno == ERANGE || optbuf == end) {
338            lcb_log(LOGARGS(obj, ERR), "Invalid value for environment LCB_SSL. (%s)", optbuf);
339            return LCB_ERR_BAD_ENVIRONMENT;
340        }
341        lcb_log(LOGARGS(obj, INFO), "SSL modified from environment. Policy is 0x%lx", env_policy);
342        settings->sslopts = env_policy;
343    }
344
345    if (settings->truststorepath == nullptr && !params.truststorepath().empty()) {
346        settings->truststorepath = lcb_strdup(params.truststorepath().c_str());
347    }
348
349    if (settings->certpath == nullptr && !params.certpath().empty()) {
350        settings->certpath = lcb_strdup(params.certpath().c_str());
351    }
352
353    if (settings->keypath == nullptr && !params.keypath().empty()) {
354        settings->keypath = lcb_strdup(params.keypath().c_str());
355    }
356
357    if (env_policy == -1) {
358        settings->sslopts = params.sslopts();
359    }
360
361    if (settings->sslopts & LCB_SSL_ENABLED) {
362        if (!(settings->sslopts & LCB_SSL_NOGLOBALINIT)) {
363            lcbio_ssl_global_init();
364        } else {
365            lcb_log(LOGARGS(obj, INFO), "ssl=no_global_init. Not initializing openssl globals");
366        }
367        if (settings->keypath && !settings->certpath) {
368            lcb_log(LOGARGS(obj, ERR), "SSL key have to be specified with certificate");
369            return LCB_ERR_INVALID_ARGUMENT;
370        }
371        settings->ssl_ctx = lcbio_ssl_new(settings->truststorepath, settings->certpath, settings->keypath,
372                                          settings->sslopts & LCB_SSL_NOVERIFY, &err, settings);
373        if (!settings->ssl_ctx) {
374            return err;
375        }
376    } else {
377        // keypath might be used to flag, that library is using SSL authentication
378        // To avoid skipping other authentication mechanisms, cleanup the keypath.
379        free(settings->keypath);
380        settings->keypath = nullptr;
381    }
382    return LCB_SUCCESS;
383}
384
385static lcb_STATUS apply_spec_options(lcb_INSTANCE *obj, const Connspec &params)
386{
387    lcb_STATUS err;
388    for (const auto &ii : params.options()) {
389        lcb_log(LOGARGS(obj, DEBUG), "Applying initial cntl %s=%s", ii.first.c_str(), ii.second.c_str());
390
391        err = lcb_cntl_string(obj, ii.first.c_str(), ii.second.c_str());
392        if (err != LCB_SUCCESS) {
393            return err;
394        }
395    }
396    return LCB_SUCCESS;
397}
398
399static lcb_STATUS apply_env_options(lcb_INSTANCE *obj)
400{
401    Connspec tmpspec;
402    const char *options = getenv("LCB_OPTIONS");
403
404    if (!options) {
405        return LCB_SUCCESS;
406    }
407
408    std::string tmp("couchbase://?");
409    tmp.append(options);
410    if (tmpspec.parse(tmp.c_str(), tmp.size()) != LCB_SUCCESS) {
411        return LCB_ERR_BAD_ENVIRONMENT;
412    }
413    return apply_spec_options(obj, tmpspec);
414}
415
416lcb_STATUS lcb_reinit(lcb_INSTANCE *obj, const char *connstr)
417{
418    Connspec params;
419    lcb_STATUS err;
420    const char *errmsg = nullptr;
421    err = params.parse(connstr, strlen(connstr), &errmsg);
422
423    if (err != LCB_SUCCESS) {
424        lcb_log(LOGARGS(obj, ERROR), "Couldn't reinit: %s", errmsg);
425    }
426
427    if (params.sslopts() != LCBT_SETTING(obj, sslopts) || !params.certpath().empty()) {
428        lcb_log(LOGARGS(obj, WARN), "Ignoring SSL reinit options");
429    }
430
431    /* apply the options */
432    err = apply_spec_options(obj, params);
433    if (err != LCB_SUCCESS) {
434        goto GT_DONE;
435    }
436    obj->populate_nodes(params);
437    err = init_providers(obj, params);
438    if (err != LCB_SUCCESS) {
439        goto GT_DONE;
440    }
441
442GT_DONE:
443    return err;
444}
445
446LIBCOUCHBASE_API
447lcb_STATUS lcb_create(lcb_INSTANCE **instance, const lcb_CREATEOPTS *options)
448{
449    Connspec spec;
450    struct lcb_io_opt_st *io_priv = nullptr;
451    lcb_INSTANCE_TYPE type = LCB_TYPE_BUCKET;
452    lcb_INSTANCE *obj = nullptr;
453    lcb_STATUS err;
454    lcb_settings *settings;
455
456    if (options) {
457        io_priv = options->io;
458        type = options->type;
459        err = spec.load(*options);
460    } else {
461        const char *errmsg;
462        const char *default_connstr = "couchbase://";
463        err = spec.parse(default_connstr, strlen(default_connstr), &errmsg);
464    }
465    if (err != LCB_SUCCESS) {
466        goto GT_DONE;
467    }
468
469    if ((obj = (lcb_INSTANCE *)calloc(1, sizeof(*obj))) == nullptr) {
470        err = LCB_ERR_NO_MEMORY;
471        goto GT_DONE;
472    }
473    obj->crypto = new std::map<std::string, lcbcrypto_PROVIDER *>();
474    obj->deferred_operations = new std::list<std::function<void(lcb_STATUS)>>();
475    if (!(settings = lcb_settings_new())) {
476        err = LCB_ERR_NO_MEMORY;
477        goto GT_DONE;
478    }
479
480    /* initialize the settings */
481    obj->settings = settings;
482    obj->settings->conntype = type;
483    obj->settings->ipv6 = spec.ipv6_policy();
484
485    if (spec.bucket().empty()) {
486        if (type == LCB_TYPE_BUCKET) {
487            settings->bucket = lcb_strdup("default");
488        }
489    } else {
490        settings->bucket = lcb_strdup(spec.bucket().c_str());
491    }
492
493    if (!spec.username().empty()) {
494        settings->auth->set_mode(LCBAUTH_MODE_RBAC);
495        err = settings->auth->add(spec.username(), spec.password(), LCBAUTH_F_CLUSTER);
496    } else {
497        if (type == LCB_TYPE_BUCKET) {
498            settings->auth->set_mode(LCBAUTH_MODE_CLASSIC);
499            err = settings->auth->add(settings->bucket, spec.password(), LCBAUTH_F_BUCKET);
500        }
501    }
502    if (err != LCB_SUCCESS) {
503        goto GT_DONE;
504    }
505
506    settings->logger = spec.logger();
507    if (settings->logger == nullptr) {
508        settings->logger = lcb_init_console_logger();
509    }
510    settings->iid = lcb_next_rand64();
511    if (spec.loglevel()) {
512        lcb_U32 val = spec.loglevel();
513        lcb_cntl(obj, LCB_CNTL_SET, LCB_CNTL_CONLOGGER_LEVEL, &val);
514    }
515    settings->log_redaction = spec.logredact();
516    if (settings->log_redaction) {
517        lcb_log(LOGARGS(obj, INFO), "Logging redaction enabled. Logs have reduced identifying information. Diagnosis "
518                                    "and support of issues may be challenging or not possible in this configuration");
519    }
520
521    lcb_log(LOGARGS(obj, INFO), "Version=%s, Changeset=%s", lcb_get_version(nullptr), LCB_VERSION_CHANGESET);
522    lcb_log(LOGARGS(obj, INFO), "Effective connection string: " LCB_LOG_SPEC("%s") ". Bucket=" LCB_LOG_SPEC("%s"),
523            settings->log_redaction ? LCB_LOG_SD_OTAG : "", spec.connstr().c_str(),
524            settings->log_redaction ? LCB_LOG_SD_CTAG : "", settings->log_redaction ? LCB_LOG_MD_OTAG : "",
525            settings->bucket, settings->log_redaction ? LCB_LOG_MD_CTAG : "");
526
527    if (io_priv == nullptr) {
528        lcb_io_opt_t ops;
529        if ((err = lcb_create_io_ops(&ops, nullptr)) != LCB_SUCCESS) {
530            goto GT_DONE;
531        }
532        io_priv = ops;
533        LCB_IOPS_BASEFLD(io_priv, need_cleanup) = 1;
534    }
535
536    obj->cmdq.cqdata = obj;
537    obj->iotable = lcbio_table_new(io_priv);
538    obj->memd_sockpool = new io::Pool(settings, obj->iotable);
539    obj->http_sockpool = new io::Pool(settings, obj->iotable);
540
541    {
542        // Needs its own scope because there are prior GOTOs
543        io::Pool::Options pool_opts;
544        pool_opts.maxidle = 1;
545        pool_opts.tmoidle = LCB_MS2US(10000); // 10 seconds
546        obj->memd_sockpool->set_options(pool_opts);
547        obj->http_sockpool->set_options(pool_opts);
548    }
549
550    obj->confmon = new clconfig::Confmon(settings, obj->iotable, obj);
551    obj->ht_nodes = new Hostlist();
552    obj->mc_nodes = new Hostlist();
553    obj->retryq = new RetryQueue(&obj->cmdq, obj->iotable, obj->settings);
554    obj->n1ql_cache = lcb_n1qlcache_create();
555    lcb_initialize_packet_handlers(obj);
556    lcb_aspend_init(&obj->pendops);
557    obj->collcache = new lcb::CollectionCache();
558
559    if ((err = setup_ssl(obj, spec)) != LCB_SUCCESS) {
560        goto GT_DONE;
561    }
562
563    if ((err = apply_spec_options(obj, spec)) != LCB_SUCCESS) {
564        goto GT_DONE;
565    }
566    if ((err = apply_env_options(obj)) != LCB_SUCCESS) {
567        goto GT_DONE;
568    }
569
570    if ((err = obj->process_dns_srv(spec)) != LCB_SUCCESS) {
571        goto GT_DONE;
572    }
573
574    obj->populate_nodes(spec);
575    if ((err = init_providers(obj, spec)) != LCB_SUCCESS) {
576        goto GT_DONE;
577    }
578    if (settings->use_tracing) {
579        if (options && options->tracer) {
580            settings->tracer = options->tracer;
581        } else {
582            settings->tracer = lcbtrace_new(obj, LCBTRACE_F_THRESHOLD);
583        }
584    }
585    if (options && options->meter) {
586        settings->meter = (new lcb::metrics::CachingMeter(options->meter))->wrap();
587#ifdef LCB_USE_HDR_HISTOGRAM
588    } else {
589        settings->meter = (new lcb::metrics::LoggingMeter(obj))->wrap();
590#endif
591    }
592
593    obj->last_error = err;
594GT_DONE:
595    if (err != LCB_SUCCESS && obj) {
596        lcb_destroy(obj);
597        *instance = nullptr;
598    } else {
599        *instance = obj;
600    }
601    return err;
602}
603
604LIBCOUCHBASE_API
605int lcb_is_redacting_logs(lcb_INSTANCE *instance)
606{
607    return instance && instance->settings && instance->settings->log_redaction;
608}
609
610typedef struct {
611    lcbio_pTABLE table;
612    lcbio_pTIMER timer;
613    int stopped;
614} SYNCDTOR;
615
616static void sync_dtor_cb(void *arg)
617{
618    auto *sd = (SYNCDTOR *)arg;
619    if (sd->table->refcount == 2) {
620        lcbio_timer_destroy(sd->timer);
621        IOT_STOP(sd->table);
622        sd->stopped = 1;
623    }
624}
625
626extern "C" {
627void lcbdur_destroy(void *);
628}
629
630static void do_pool_shutdown(io::Pool *pool)
631{
632    pool->shutdown();
633}
634
635LIBCOUCHBASE_API
636void lcb_destroy(lcb_INSTANCE *instance)
637{
638    instance->destroying = 1;
639#define DESTROY(fn, fld)                                                                                               \
640    if (instance->fld) {                                                                                               \
641        fn(instance->fld);                                                                                             \
642        instance->fld = nullptr;                                                                                       \
643    }
644
645    lcb_ASPEND *po = &instance->pendops;
646    lcb_ASPEND_SETTYPE::iterator it;
647    lcb_ASPEND_SETTYPE *pendq;
648
649    DESTROY(delete, bs_state)
650    DESTROY(delete, ht_nodes)
651    DESTROY(delete, mc_nodes)
652
653    lcb::cancel_deferred_operations(instance);
654    delete instance->deferred_operations;
655
656    if ((pendq = po->items[LCB_PENDTYPE_DURABILITY])) {
657        std::vector<void *> dsets(pendq->begin(), pendq->end());
658        for (auto &dset : dsets) {
659            lcbdur_destroy(dset);
660        }
661        pendq->clear();
662    }
663
664    for (size_t ii = 0; ii < LCBT_NSERVERS(instance); ++ii) {
665        instance->get_server(ii)->close();
666    }
667
668    if ((pendq = po->items[LCB_PENDTYPE_HTTP])) {
669        std::vector<void *> requests(pendq->begin(), pendq->end());
670        for (void *request : requests) {
671            auto *htreq = reinterpret_cast<http::Request *>(request);
672            htreq->finish(LCB_ERR_REQUEST_CANCELED);
673        }
674    }
675
676    DESTROY(delete, retryq)
677    DESTROY(delete, confmon)
678    DESTROY(do_pool_shutdown, memd_sockpool)
679    DESTROY(do_pool_shutdown, http_sockpool)
680    DESTROY(lcb_vbguess_destroy, vbguess)
681    DESTROY(lcb_n1qlcache_destroy, n1ql_cache)
682
683    if (instance->cmdq.pipelines) {
684        unsigned ii;
685        for (ii = 0; ii < instance->cmdq.npipelines; ii++) {
686            auto *server = static_cast<lcb::Server *>(instance->cmdq.pipelines[ii]);
687            if (server) {
688                server->instance = nullptr;
689                server->parent = nullptr;
690            }
691        }
692    }
693    mcreq_queue_cleanup(&instance->cmdq);
694    DESTROY(delete, collcache)
695    if (instance->cur_configinfo) {
696        instance->cur_configinfo->decref();
697        instance->cur_configinfo = nullptr;
698    }
699    instance->cmdq.config = nullptr;
700    instance->cmdq.cqdata = nullptr;
701    lcb_aspend_cleanup(po);
702
703    if (instance->settings && instance->settings->tracer) {
704        lcbtrace_destroy(instance->settings->tracer);
705        instance->settings->tracer = nullptr;
706    }
707
708    if (instance->iotable && instance->iotable->refcount > 1 && instance->settings && instance->settings->syncdtor) {
709        /* create an async object */
710        SYNCDTOR sd;
711        sd.table = instance->iotable;
712        sd.timer = lcbio_timer_new(instance->iotable, &sd, sync_dtor_cb);
713        sd.stopped = 0;
714        lcbio_async_signal(sd.timer);
715        lcb_log(LOGARGS(instance, WARN), "Running event loop to drain any pending I/O events");
716        do {
717            IOT_START(instance->iotable);
718        } while (!sd.stopped);
719    }
720
721    // Once we are done destroying the instance, we need to manually disconnect
722    // the logger from the settings since further work may proceed in the background
723    // with some forms of IO backend, but once lcb_destroy is invoked, the logger
724    // may no longer be valid from the application side.
725    if (instance->settings && instance->settings->logger) {
726        instance->settings->logger = nullptr;
727    }
728
729    DESTROY(lcbio_table_unref, iotable)
730    DESTROY(lcb_settings_unref, settings)
731    DESTROY(lcb_histogram_destroy, kv_timings)
732    if (instance->scratch) {
733        delete instance->scratch;
734        instance->scratch = nullptr;
735    }
736
737    for (auto &ii : *instance->crypto) {
738        lcbcrypto_unref(ii.second);
739    }
740    delete instance->crypto;
741    instance->crypto = nullptr;
742
743    delete[] instance->dcpinfo;
744    memset(instance, 0xff, sizeof(*instance));
745    free(instance);
746#undef DESTROY
747}
748
749static void destroy_cb(void *arg)
750{
751    auto *instance = static_cast<lcb_INSTANCE *>(arg);
752    lcbio_timer_destroy(instance->dtor_timer);
753    lcb_destroy(instance);
754}
755
756LIBCOUCHBASE_API
757void lcb_destroy_async(lcb_INSTANCE *instance, const void *arg)
758{
759    instance->dtor_timer = lcbio_timer_new(instance->iotable, instance, destroy_cb);
760    instance->settings->dtorarg = (void *)arg;
761    lcbio_async_signal(instance->dtor_timer);
762}
763
764lcb::Server *lcb_st::find_server(const lcb_host_t &host) const
765{
766    unsigned ii;
767    for (ii = 0; ii < cmdq.npipelines; ii++) {
768        auto *server = static_cast<lcb::Server *>(cmdq.pipelines[ii]);
769        if (server && server->has_valid_host() && lcb_host_equals(&server->get_host(), &host)) {
770            return server;
771        }
772    }
773    return nullptr;
774}
775
776LIBCOUCHBASE_API
777lcb_STATUS lcb_connect(lcb_INSTANCE *instance)
778{
779    return instance->bootstrap(BS_REFRESH_INITIAL);
780}
781
782LIBCOUCHBASE_API
783lcb_STATUS lcb_open(lcb_INSTANCE *instance, const char *bucket, size_t bucket_len)
784{
785    if (bucket == nullptr) {
786        lcb_log(LOGARGS(instance, ERR), "Bucket name cannot be a nullptr, sorry");
787        return LCB_ERR_INVALID_ARGUMENT;
788    }
789    lcbvb_CONFIG *cfg = LCBT_VBCONFIG(instance);
790    if (cfg == nullptr) {
791        lcb_log(LOGARGS(instance, ERR),
792                "The instance wasn't not bootstrapped, unable to associate it with bucket, sorry");
793        return LCB_ERR_INVALID_ARGUMENT;
794    }
795    if (LCBVB_BUCKET_NAME(cfg)) {
796        lcb_log(LOGARGS(instance, ERR), "The instance has been associated with the bucket already, sorry");
797        return LCB_ERR_INVALID_ARGUMENT;
798    }
799    instance->settings->conntype = LCB_TYPE_BUCKET;
800    instance->settings->bucket = (char *)calloc(bucket_len + 1, sizeof(char));
801    memcpy(instance->settings->bucket, bucket, bucket_len);
802    for (unsigned ii = 0; ii < instance->cmdq.npipelines; ii++) {
803        auto *server = static_cast<lcb::Server *>(instance->cmdq.pipelines[ii]);
804        if (!server->selected_bucket && server->connctx) {
805            lcb::MemcachedRequest req(PROTOCOL_BINARY_CMD_SELECT_BUCKET);
806            req.opaque(0xcafe);
807            req.sizes(0, bucket_len, 0);
808            lcbio_ctx_put(server->connctx, req.data(), req.size());
809            server->bucket.assign(bucket, bucket_len);
810            lcbio_ctx_put(server->connctx, bucket, bucket_len);
811            server->flush();
812        }
813    }
814
815    return instance->bootstrap(BS_REFRESH_OPEN_BUCKET);
816}
817
818LIBCOUCHBASE_API
819void *lcb_mem_alloc(lcb_size_t size)
820{
821    return malloc(size);
822}
823
824LIBCOUCHBASE_API
825void lcb_mem_free(void *ptr)
826{
827    free(ptr);
828}
829
830LCB_INTERNAL_API
831void lcb_run_loop(lcb_INSTANCE *instance)
832{
833    IOT_START(instance->iotable);
834}
835
836LCB_INTERNAL_API
837void lcb_stop_loop(lcb_INSTANCE *instance)
838{
839    IOT_STOP(instance->iotable);
840}
841
842void lcb_aspend_init(lcb_ASPEND *ops)
843{
844    unsigned ii;
845    for (ii = 0; ii < LCB_PENDTYPE_MAX; ++ii) {
846        ops->items[ii] = new lcb_ASPEND_SETTYPE();
847    }
848    ops->count = 0;
849}
850
851void lcb_aspend_add(lcb_ASPEND *ops, lcb_ASPENDTYPE type, const void *item)
852{
853    ops->count++;
854    if (type == LCB_PENDTYPE_COUNTER) {
855        return;
856    }
857    ops->items[type]->insert(const_cast<void *>(item));
858}
859
860void lcb_aspend_del(lcb_ASPEND *ops, lcb_ASPENDTYPE type, const void *item)
861{
862    if (type == LCB_PENDTYPE_COUNTER) {
863        ops->count--;
864        return;
865    }
866    if (ops->items[type]->erase(const_cast<void *>(item)) != 0) {
867        ops->count--;
868    }
869}
870
871void lcb_aspend_cleanup(lcb_ASPEND *ops)
872{
873    unsigned ii;
874    for (ii = 0; ii < LCB_PENDTYPE_MAX; ii++) {
875        delete ops->items[ii];
876    }
877}
878
879LIBCOUCHBASE_API
880void lcb_sched_enter(lcb_INSTANCE *instance)
881{
882    mcreq_sched_enter(&instance->cmdq);
883}
884LIBCOUCHBASE_API
885void lcb_sched_leave(lcb_INSTANCE *instance)
886{
887    mcreq_sched_leave(&instance->cmdq, LCBT_SETTING(instance, sched_implicit_flush));
888}
889LIBCOUCHBASE_API
890void lcb_sched_fail(lcb_INSTANCE *instance)
891{
892    mcreq_sched_fail(&instance->cmdq);
893}
894
895LIBCOUCHBASE_API
896int lcb_supports_feature(int n)
897{
898    if (n == LCB_SUPPORTS_TRACING) {
899        return 1;
900    }
901    if (n == LCB_SUPPORTS_SNAPPY) {
902        return 1;
903    }
904    if (n == LCB_SUPPORTS_SSL) {
905        return lcbio_ssl_supported();
906    } else {
907        return 0;
908    }
909}
910
911LCB_INTERNAL_API void lcb_loop_ref(lcb_INSTANCE *instance)
912{
913    lcb_aspend_add(&instance->pendops, LCB_PENDTYPE_COUNTER, nullptr);
914}
915LCB_INTERNAL_API void lcb_loop_unref(lcb_INSTANCE *instance)
916{
917    lcb_aspend_del(&instance->pendops, LCB_PENDTYPE_COUNTER, nullptr);
918    lcb_maybe_breakout(instance);
919}
920
921LCB_INTERNAL_API uint32_t lcb_durability_timeout(lcb_INSTANCE *instance, uint32_t tmo_us)
922{
923    if (tmo_us == 0) {
924        tmo_us = instance->settings->operation_timeout;
925    }
926    if (tmo_us < instance->settings->persistence_timeout_floor) {
927        lcb_log(LOGARGS(instance, WARN), "Durability timeout is too low (%uus), using %uus instead", tmo_us,
928                instance->settings->persistence_timeout_floor);
929        tmo_us = instance->settings->persistence_timeout_floor;
930    }
931    return tmo_us / 1000 * 0.9;
932}
933
934static bool is_valid_collection_char(char ch)
935{
936    if (ch >= 'A' && ch <= 'Z') {
937        return true;
938    }
939    if (ch >= 'a' && ch <= 'z') {
940        return true;
941    }
942    if (ch >= '0' && ch <= '9') {
943        return true;
944    }
945    switch (ch) {
946        case '_':
947        case '-':
948        case '%':
949            return true;
950        default:
951            return false;
952    }
953}
954
955static bool is_valid_collection_element(const char *element, size_t element_len)
956{
957    if (element_len == 0 || element == nullptr) {
958        /* nullptr/0 for collection is mapped to default collection */
959        return true;
960    }
961    if (element_len < 1 || element_len > 30 || element == nullptr) {
962        return false;
963    }
964    for (size_t i = 0; i < element_len; ++i) {
965        if (!is_valid_collection_char(element[i])) {
966            return false;
967        }
968    }
969    return true;
970}
971
972static bool is_default_collection_element(const char *element, size_t element_len)
973{
974    static const std::string default_name("_default");
975    if (element_len == 0 || element == nullptr || default_name.compare(0, element_len, element) == 0) {
976        return true;
977    }
978    return false;
979}
980
981LCB_INTERNAL_API lcb_STATUS lcb_is_collection_valid(lcb_INSTANCE *instance, const char *scope, size_t scope_len,
982                                                    const char *collection, size_t collection_len)
983{
984    if (!LCBT_SETTING(instance, use_collections) && !(is_default_collection_element(scope, scope_len) &&
985                                                      is_default_collection_element(collection, collection_len))) {
986        /* only allow default collection when collections disabled for the instance */
987        return LCB_ERR_SDK_FEATURE_UNAVAILABLE;
988    }
989    if (is_valid_collection_element(scope, scope_len) && is_valid_collection_element(collection, collection_len)) {
990        return LCB_SUCCESS;
991    }
992    return LCB_ERR_INVALID_ARGUMENT;
993}
994
995LCB_INTERNAL_API lcb_STATUS lcb_is_collection_valid(lcb_INSTANCE *instance, const std::string &scope,
996                                                    const std::string &collection)
997{
998    return lcb_is_collection_valid(instance, scope.c_str(), scope.size(), collection.c_str(), collection.size());
999}
1000
1001LIBCOUCHBASE_API
1002lcb_STATUS lcb_enable_timings(lcb_INSTANCE *instance)
1003{
1004    if (instance->kv_timings != nullptr) {
1005        return LCB_ERR_DOCUMENT_EXISTS;
1006    }
1007    instance->kv_timings = lcb_histogram_create();
1008    return instance->kv_timings == nullptr ? LCB_ERR_NO_MEMORY : LCB_SUCCESS;
1009}
1010
1011LIBCOUCHBASE_API
1012lcb_STATUS lcb_disable_timings(lcb_INSTANCE *instance)
1013{
1014    if (instance->kv_timings == nullptr) {
1015        return LCB_ERR_DOCUMENT_NOT_FOUND;
1016    }
1017    lcb_histogram_destroy(instance->kv_timings);
1018    instance->kv_timings = nullptr;
1019    return LCB_SUCCESS;
1020}
1021
1022typedef struct {
1023    lcb_INSTANCE *instance;
1024    const void *real_cookie;
1025    lcb_timings_callback real_cb;
1026} timings_wrapper;
1027
1028static void timings_wrapper_callback(const void *cookie, lcb_timeunit_t unit, lcb_U32 start, lcb_U32 end, lcb_U32 val,
1029                                     lcb_U32 max)
1030{
1031    const auto *wrap = (const timings_wrapper *)cookie;
1032    wrap->real_cb(wrap->instance, wrap->real_cookie, unit, start, end, val, max);
1033}
1034
1035LIBCOUCHBASE_API
1036lcb_STATUS lcb_get_timings(lcb_INSTANCE *instance, const void *cookie, lcb_timings_callback cb)
1037{
1038    timings_wrapper wrap;
1039    wrap.instance = instance;
1040    wrap.real_cookie = cookie;
1041    wrap.real_cb = cb;
1042
1043    if (!instance->kv_timings) {
1044        return LCB_ERR_DOCUMENT_NOT_FOUND;
1045    }
1046    lcb_histogram_read(instance->kv_timings, &wrap, timings_wrapper_callback);
1047    return LCB_SUCCESS;
1048}
1049
1050LCB_INTERNAL_API
1051const char *lcb_strerror_short(lcb_STATUS error)
1052{
1053#define X(c, v, t, f, s)                                                                                               \
1054    if (error == c) {                                                                                                  \
1055        return #c " (" #v ")";                                                                                         \
1056    }
1057    LCB_XERROR(X)
1058#undef X
1059    return "<FIXME: Not an LCB error>";
1060}
1061
1062LCB_INTERNAL_API
1063const char *lcb_strerror_long(lcb_STATUS error)
1064{
1065#define X(c, v, t, f, s)                                                                                               \
1066    if (error == c) {                                                                                                  \
1067        return #c " (" #v "): " s;                                                                                     \
1068    }
1069    LCB_XERROR(X)
1070#undef X
1071    return "<FIXME: Not an LCB error>";
1072}
1073
1074LIBCOUCHBASE_API
1075uint32_t lcb_error_flags(lcb_STATUS err)
1076{
1077#define X(c, v, t, f, s)                                                                                               \
1078    if (err == c) {                                                                                                    \
1079        return (uint32_t)f;                                                                                            \
1080    }
1081    LCB_XERROR(X)
1082#undef X
1083    return 0;
1084}
1085