xapian-core/backends/remote/remote-database.cc

   1 /** @file
   2  *  @brief Remote backend database class
   3  */
   4 /* Copyright (C) 2006-2024 Olly Betts
   5  * Copyright (C) 2007,2009,2010 Lemur Consulting Ltd
   6  *
   7  * This program is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU General Public License as
   9  * published by the Free Software Foundation; either version 2 of the
  10  * License, or (at your option) any later version.
  11  *
  12  * This program is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  * GNU General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU General Public License
  18  * along with this program; if not, write to the Free Software
  19  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  20  */
  21
  22 #include <config.h>
  23
  24 #include "remote-database.h"
  25
  26 #include <signal.h>
  27 #include "safesyssocket.h" // For MSG_NOSIGNAL.
  28
  29 #include "api/msetinternal.h"
  30 #include "api/smallvector.h"
  31 #include "backends/contiguousalldocspostlist.h"
  32 #include "backends/inmemory/inmemory_positionlist.h"
  33 #include "net_postlist.h"
  34 #include "remote-document.h"
  35 #include "omassert.h"
  36 #include "realtime.h"
  37 #include "net/serialise.h"
  38 #include "net/serialise-error.h"
  39 #include "pack.h"
  40 #include "remote_alltermslist.h"
  41 #include "remote_keylist.h"
  42 #include "remote_termlist.h"
  43 #include "serialise-double.h"
  44 #include "str.h"
  45 #include "stringutils.h" // For STRINGIZE().
  46 #include "weight/weightinternal.h"
  47
  48 #include <cerrno>
  49 #include <memory>
  50 #include <string>
  51 #include <string_view>
  52 #include <vector>
  53
  54 #include "xapian/constants.h"
  55 #include "xapian/error.h"
  56 #include "xapian/matchspy.h"
  57
  58 using namespace std;
  59 using Xapian::Internal::intrusive_ptr;
  60
  61 /// Return true if further replies should be expected.
  62 static inline bool
  63 is_intermediate_reply(int reply_code)
  64 {
  65     return reply_code == REPLY_DOCDATA ||
  66            reply_code == REPLY_VALUE ||
  67            reply_code == REPLY_TERMLISTHEADER ||
  68            reply_code == REPLY_POSTLISTHEADER;
  69 }
  70
  71 [[noreturn]]
  72 static void
  73 throw_invalid_operation(const char* message)
  74 {
  75     throw Xapian::InvalidOperationError(message);
  76 }
  77
  78 [[noreturn]]
  79 static void
  80 throw_handshake_failed(const string & context)
  81 {
  82     throw Xapian::NetworkError("Handshake failed - is this a Xapian server?",
  83                                context);
  84 }
  85
  86 [[noreturn]]
  87 static void
  88 throw_connection_closed_unexpectedly()
  89 {
  90     throw Xapian::NetworkError("Connection closed unexpectedly");
  91 }
  92
  93 RemoteDatabase::RemoteDatabase(pair<int, string> fd_and_context,
  94                                double timeout_,
  95                                bool writable,
  96                                int flags)
  97     : Xapian::Database::Internal(writable ?
  98                                  TRANSACTION_NONE :
  99                                  TRANSACTION_READONLY),
 100       link(fd_and_context.first, fd_and_context.first, fd_and_context.second),
 101       cached_stats_valid(),
 102       mru_valstats(),
 103       mru_slot(Xapian::BAD_VALUENO),
 104       timeout(timeout_)
 105 {
 106     update_stats(MSG_MAX);
 107
 108     if (writable) {
 109         if (flags & Xapian::DB_RETRY_LOCK) {
 110             string message;
 111             pack_uint_last(message, unsigned(flags & Xapian::DB_RETRY_LOCK));
 112             update_stats(MSG_WRITEACCESS, message);
 113         } else {
 114             update_stats(MSG_WRITEACCESS);
 115         }
 116     }
 117 }
 118
 119 Xapian::termcount
 120 RemoteDatabase::positionlist_count(Xapian::docid did,
 121                                    std::string_view term) const
 122 {
 123     if (cached_stats_valid && !has_positional_info)
 124         return 0;
 125
 126     string message;
 127     pack_uint(message, did);
 128     message += term;
 129     send_message(MSG_POSITIONLISTCOUNT, message);
 130
 131     get_message(message, REPLY_POSITIONLISTCOUNT);
 132     const char * p = message.data();
 133     const char * p_end = p + message.size();
 134     Xapian::termcount count;
 135     if (!unpack_uint_last(&p, p_end, &count)) {
 136         throw Xapian::NetworkError("Bad REPLY_POSITIONLISTCOUNT",
 137                                    link.get_context());
 138     }
 139     return count;
 140 }
 141
 142 void
 143 RemoteDatabase::keep_alive()
 144 {
 145     send_message(MSG_KEEPALIVE, {});
 146     string message;
 147     get_message(message, REPLY_DONE);
 148 }
 149
 150 TermList*
 151 RemoteDatabase::open_metadata_keylist(std::string_view prefix) const
 152 {
 153     send_message(MSG_METADATAKEYLIST, prefix);
 154     string message;
 155     get_message(message, REPLY_METADATAKEYLIST);
 156     return new RemoteKeyList(prefix, std::move(message));
 157 }
 158
 159 TermList *
 160 RemoteDatabase::open_term_list(Xapian::docid did) const
 161 {
 162     Assert(did);
 163
 164     // Ensure that total_length and doccount are up-to-date.
 165     if (!cached_stats_valid) update_stats();
 166
 167     string message;
 168     pack_uint_last(message, did);
 169     send_message(MSG_TERMLIST, message);
 170
 171     get_message(message, REPLY_TERMLISTHEADER);
 172     const char * p = message.c_str();
 173     const char * p_end = p + message.size();
 174     Xapian::termcount doclen;
 175     Xapian::termcount num_entries;
 176     if (!unpack_uint(&p, p_end, &doclen) ||
 177         !unpack_uint_last(&p, p_end, &num_entries)) {
 178         throw Xapian::NetworkError("Bad REPLY_TERMLISTHEADER",
 179                                    link.get_context());
 180     }
 181     get_message(message, REPLY_TERMLIST);
 182     return new RemoteTermList(num_entries, doclen, doccount, this, did,
 183                               std::move(message));
 184 }
 185
 186 TermList *
 187 RemoteDatabase::open_term_list_direct(Xapian::docid did) const
 188 {
 189     return RemoteDatabase::open_term_list(did);
 190 }
 191
 192 TermList*
 193 RemoteDatabase::open_allterms(string_view prefix) const
 194 {
 195     send_message(MSG_ALLTERMS, prefix);
 196     string message;
 197     get_message(message, REPLY_ALLTERMS);
 198     return new RemoteAllTermsList(prefix, std::move(message));
 199 }
 200
 201 PostList*
 202 RemoteDatabase::open_post_list(string_view term) const
 203 {
 204     if (term.empty()) {
 205         if (!cached_stats_valid) update_stats();
 206         if (rare(doccount == 0))
 207             return nullptr;
 208         if (doccount == lastdocid) {
 209             // The used docid range is exactly 1 to doccount inclusive.
 210             return new ContiguousAllDocsPostList(doccount);
 211         }
 212     }
 213
 214     send_message(MSG_POSTLIST, term);
 215
 216     string message;
 217     get_message(message, REPLY_POSTLISTHEADER);
 218
 219     const char * p = message.data();
 220     const char * p_end = p + message.size();
 221     Xapian::doccount termfreq;
 222     if (!unpack_uint_last(&p, p_end, &termfreq)) {
 223         unpack_throw_serialisation_error(p);
 224     }
 225
 226     get_message(message, REPLY_POSTLIST);
 227
 228     return new NetworkPostList(intrusive_ptr<const RemoteDatabase>(this),
 229                                term,
 230                                termfreq,
 231                                std::move(message));
 232 }
 233
 234 LeafPostList*
 235 RemoteDatabase::open_leaf_post_list(string_view, bool) const
 236 {
 237     // This method is only called during the match, and remote shards are
 238     // handled by running the match on the remote.
 239     Assert(false);
 240     return nullptr;
 241 }
 242
 243 PositionList*
 244 RemoteDatabase::open_position_list(Xapian::docid did, string_view term) const
 245 {
 246     string message;
 247     pack_uint(message, did);
 248     message += term;
 249     send_message(MSG_POSITIONLIST, message);
 250
 251     get_message(message, REPLY_POSITIONLIST);
 252     if (message.empty())
 253         return nullptr;
 254
 255     Xapian::VecCOW<Xapian::termpos> positions;
 256     Xapian::termpos lastpos = static_cast<Xapian::termpos>(-1);
 257     const char* p = message.data();
 258     const char* p_end = p + message.size();
 259     while (p != p_end) {
 260         Xapian::termpos inc;
 261         if (!unpack_uint(&p, p_end, &inc)) {
 262             unpack_throw_serialisation_error(p);
 263         }
 264         UNSIGNED_OVERFLOW_OK(lastpos += inc + 1);
 265         positions.push_back(lastpos);
 266     }
 267
 268     return new InMemoryPositionList(std::move(positions));
 269 }
 270
 271 bool
 272 RemoteDatabase::has_positions() const
 273 {
 274     if (!cached_stats_valid) update_stats();
 275     return has_positional_info;
 276 }
 277
 278 bool
 279 RemoteDatabase::reopen()
 280 {
 281     mru_slot = Xapian::BAD_VALUENO;
 282     return update_stats(MSG_REOPEN);
 283 }
 284
 285 void
 286 RemoteDatabase::close()
 287 {
 288     do_close();
 289 }
 290
 291 // Currently lazy is used:
 292 //
 293 // * To implement API flag Xapian::DOC_ASSUME_VALID which can be specified when
 294 //   calling method Database::get_document()
 295 //
 296 // * To read values for backends without streamed values in SlowValueList
 297 //
 298 // * If you call get_data(), values_begin() or values_count() on a Document
 299 //   object passed to a KeyMaker, MatchDecider, MatchSpy during the match
 300 //
 301 // The first is relevant to the remote backend, but doesn't happen during
 302 // the match.
 303 //
 304 // SlowValueList is used with the remote backend, but not to read values
 305 // during the match.
 306 //
 307 // KeyMaker and MatchSpy happens on the server with the remote backend, so
 308 // they aren't relevant here.
 309 //
 310 // So the cases which are relevant to the remote backend don't matter during
 311 // the match, and so we can ignore the lazy flag here without affecting matcher
 312 // performance.
 313 Xapian::Document::Internal *
 314 RemoteDatabase::open_document(Xapian::docid did, bool /*lazy*/) const
 315 {
 316     Assert(did);
 317
 318     string message;
 319     pack_uint_last(message, did);
 320     send_message(MSG_DOCUMENT, message);
 321
 322     string doc_data;
 323     get_message(doc_data, REPLY_DOCDATA);
 324
 325     map<Xapian::valueno, string> values;
 326     while (get_message_or_done(message, REPLY_VALUE)) {
 327         const char * p = message.data();
 328         const char * p_end = p + message.size();
 329         Xapian::valueno slot;
 330         if (!unpack_uint(&p, p_end, &slot)) {
 331             unpack_throw_serialisation_error(p);
 332         }
 333         values.insert(make_pair(slot, string(p, p_end)));
 334     }
 335
 336     return new RemoteDocument(this, did, std::move(doc_data),
 337                               std::move(values));
 338 }
 339
 340 bool
 341 RemoteDatabase::update_stats(message_type msg_code, const string & body) const
 342 {
 343     // MSG_MAX signals that we're handling the opening greeting, which isn't in
 344     // response to an explicit message.
 345     if (msg_code != MSG_MAX)
 346         send_message(msg_code, body);
 347
 348     string message;
 349     if (!get_message_or_done(message, REPLY_UPDATE)) {
 350         // The database was already open at the latest revision.
 351         return false;
 352     }
 353
 354     if (message.size() < 3) {
 355         throw_handshake_failed(link.get_context());
 356     }
 357     const char *p = message.c_str();
 358     const char *p_end = p + message.size();
 359
 360     // The protocol major versions must match.  The protocol minor version of
 361     // the server must be >= that of the client.
 362     int protocol_major = static_cast<unsigned char>(*p++);
 363     int protocol_minor = static_cast<unsigned char>(*p++);
 364     if (protocol_major != XAPIAN_REMOTE_PROTOCOL_MAJOR_VERSION ||
 365         protocol_minor < XAPIAN_REMOTE_PROTOCOL_MINOR_VERSION) {
 366         string errmsg("Server supports protocol version");
 367         if (protocol_minor) {
 368             errmsg += "s ";
 369             errmsg += str(protocol_major);
 370             errmsg += ".0 to ";
 371         } else {
 372             errmsg += ' ';
 373         }
 374         errmsg += str(protocol_major);
 375         errmsg += '.';
 376         errmsg += str(protocol_minor);
 377         errmsg +=
 378             " - client is using "
 379             STRINGIZE(XAPIAN_REMOTE_PROTOCOL_MAJOR_VERSION)
 380             "."
 381             STRINGIZE(XAPIAN_REMOTE_PROTOCOL_MINOR_VERSION);
 382         throw Xapian::NetworkError(errmsg, link.get_context());
 383     }
 384
 385     if (!unpack_uint(&p, p_end, &doccount) ||
 386         !unpack_uint(&p, p_end, &lastdocid) ||
 387         !unpack_uint(&p, p_end, &doclen_lbound) ||
 388         !unpack_uint(&p, p_end, &doclen_ubound) ||
 389         !unpack_bool(&p, p_end, &has_positional_info) ||
 390         !unpack_uint(&p, p_end, &total_length)) {
 391         throw Xapian::NetworkError("Bad stats update message received",
 392                                    link.get_context());
 393     }
 394     lastdocid += doccount;
 395     doclen_ubound += doclen_lbound;
 396     uuid.assign(p, p_end);
 397     cached_stats_valid = true;
 398     return true;
 399 }
 400
 401 Xapian::doccount
 402 RemoteDatabase::get_doccount() const
 403 {
 404     if (!cached_stats_valid) update_stats();
 405     return doccount;
 406 }
 407
 408 Xapian::docid
 409 RemoteDatabase::get_lastdocid() const
 410 {
 411     if (!cached_stats_valid) update_stats();
 412     return lastdocid;
 413 }
 414
 415 Xapian::totallength
 416 RemoteDatabase::get_total_length() const
 417 {
 418     if (!cached_stats_valid) update_stats();
 419     return total_length;
 420 }
 421
 422 bool
 423 RemoteDatabase::term_exists(string_view term) const
 424 {
 425     if (term.empty()) {
 426         return get_doccount() != 0;
 427     }
 428     send_message(MSG_TERMEXISTS, term);
 429     string message;
 430     reply_type type = get_message(message,
 431                                   REPLY_TERMEXISTS,
 432                                   REPLY_TERMDOESNTEXIST);
 433     return (type == REPLY_TERMEXISTS);
 434 }
 435
 436 void
 437 RemoteDatabase::get_freqs(string_view term,
 438                           Xapian::doccount* termfreq_ptr,
 439                           Xapian::termcount* collfreq_ptr) const
 440 {
 441     Assert(!term.empty());
 442     string message;
 443     if (termfreq_ptr && collfreq_ptr) {
 444         send_message(MSG_FREQS, term);
 445         get_message(message, REPLY_FREQS);
 446         const char* p = message.data();
 447         const char* p_end = p + message.size();
 448         if (unpack_uint(&p, p_end, termfreq_ptr) &&
 449             unpack_uint_last(&p, p_end, collfreq_ptr)) {
 450             return;
 451         }
 452     } else if (termfreq_ptr) {
 453         send_message(MSG_TERMFREQ, term);
 454         get_message(message, REPLY_TERMFREQ);
 455         const char* p = message.data();
 456         const char* p_end = p + message.size();
 457         if (unpack_uint_last(&p, p_end, termfreq_ptr)) {
 458             return;
 459         }
 460     } else if (collfreq_ptr) {
 461         send_message(MSG_COLLFREQ, term);
 462         get_message(message, REPLY_COLLFREQ);
 463         const char* p = message.data();
 464         const char* p_end = p + message.size();
 465         if (unpack_uint_last(&p, p_end, collfreq_ptr)) {
 466             return;
 467         }
 468     } else {
 469         Assert(false);
 470         return;
 471     }
 472     throw Xapian::NetworkError("Bad REPLY_FREQS/REPLY_TERMFREQ/REPLY_COLLFREQ",
 473                                link.get_context());
 474 }
 475
 476 void
 477 RemoteDatabase::read_value_stats(Xapian::valueno slot) const
 478 {
 479     if (mru_slot == slot)
 480         return;
 481
 482     string message;
 483     pack_uint_last(message, slot);
 484     send_message(MSG_VALUESTATS, message);
 485
 486     get_message(message, REPLY_VALUESTATS);
 487     const char* p = message.data();
 488     const char* p_end = p + message.size();
 489     mru_slot = slot;
 490     if (!unpack_uint(&p, p_end, &mru_valstats.freq) ||
 491         !unpack_string(&p, p_end, mru_valstats.lower_bound)) {
 492         throw Xapian::NetworkError("Bad REPLY_VALUESTATS", link.get_context());
 493     }
 494     mru_valstats.upper_bound.assign(p, p_end);
 495 }
 496
 497 Xapian::doccount
 498 RemoteDatabase::get_value_freq(Xapian::valueno slot) const
 499 {
 500     read_value_stats(slot);
 501     return mru_valstats.freq;
 502 }
 503
 504 std::string
 505 RemoteDatabase::get_value_lower_bound(Xapian::valueno slot) const
 506 {
 507     read_value_stats(slot);
 508     return mru_valstats.lower_bound;
 509 }
 510
 511 std::string
 512 RemoteDatabase::get_value_upper_bound(Xapian::valueno slot) const
 513 {
 514     read_value_stats(slot);
 515     return mru_valstats.upper_bound;
 516 }
 517
 518 Xapian::termcount
 519 RemoteDatabase::get_doclength_lower_bound() const
 520 {
 521     return doclen_lbound;
 522 }
 523
 524 Xapian::termcount
 525 RemoteDatabase::get_doclength_upper_bound() const
 526 {
 527     return doclen_ubound;
 528 }
 529
 530 Xapian::termcount
 531 RemoteDatabase::get_wdf_upper_bound(string_view) const
 532 {
 533     // The default implementation returns get_collection_freq(), but we
 534     // don't want the overhead of a remote message and reply per query
 535     // term, and we can get called in the middle of a remote exchange
 536     // too.  FIXME: handle this bound in the stats local/remote code...
 537     return doclen_ubound;
 538 }
 539
 540 Xapian::termcount
 541 RemoteDatabase::get_doclength(Xapian::docid did) const
 542 {
 543     Assert(did != 0);
 544     string message;
 545     pack_uint_last(message, did);
 546     send_message(MSG_DOCLENGTH, message);
 547
 548     get_message(message, REPLY_DOCLENGTH);
 549     const char* p = message.c_str();
 550     const char* p_end = p + message.size();
 551     Xapian::termcount doclen;
 552     if (!unpack_uint_last(&p, p_end, &doclen)) {
 553         throw Xapian::NetworkError("Bad REPLY_DOCLENGTH", link.get_context());
 554     }
 555     return doclen;
 556 }
 557
 558 Xapian::termcount
 559 RemoteDatabase::get_unique_terms(Xapian::docid did) const
 560 {
 561     Assert(did != 0);
 562     string message;
 563     pack_uint_last(message, did);
 564     send_message(MSG_UNIQUETERMS, message);
 565
 566     get_message(message, REPLY_UNIQUETERMS);
 567     const char* p = message.c_str();
 568     const char* p_end = p + message.size();
 569     Xapian::termcount doclen;
 570     if (!unpack_uint_last(&p, p_end, &doclen)) {
 571         throw Xapian::NetworkError("Bad REPLY_DOCLENGTH", link.get_context());
 572     }
 573     return doclen;
 574 }
 575
 576 Xapian::termcount
 577 RemoteDatabase::get_wdfdocmax(Xapian::docid did) const
 578 {
 579     Assert(did != 0);
 580     string message;
 581     pack_uint_last(message, did);
 582     send_message(MSG_WDFDOCMAX, message);
 583
 584     get_message(message, REPLY_WDFDOCMAX);
 585     const char* p = message.c_str();
 586     const char* p_end = p + message.size();
 587     Xapian::termcount wdfdocmax;
 588     if (!unpack_uint_last(&p, p_end, &wdfdocmax)) {
 589         throw Xapian::NetworkError("Bad REPLY_WDFDOCMAX", link.get_context());
 590     }
 591     return wdfdocmax;
 592 }
 593
 594 reply_type
 595 RemoteDatabase::get_message(string &result,
 596                             reply_type required_type,
 597                             reply_type required_type2) const
 598 {
 599     double end_time = RealTime::end_time(timeout);
 600     int type = link.get_message(result, end_time);
 601     if (pending_reply && !is_intermediate_reply(type)) {
 602         pending_reply = false;
 603     }
 604     if (type < 0)
 605         throw_connection_closed_unexpectedly();
 606     if (rare(type) >= REPLY_MAX) {
 607         if (required_type == REPLY_UPDATE)
 608             throw_handshake_failed(link.get_context());
 609         string errmsg("Invalid reply type ");
 610         errmsg += str(type);
 611         throw Xapian::NetworkError(errmsg);
 612     }
 613     if (type == REPLY_EXCEPTION) {
 614         unserialise_error(result, "REMOTE:", link.get_context());
 615     }
 616     if (type != required_type && type != required_type2) {
 617         string errmsg("Expecting reply type ");
 618         errmsg += str(int(required_type));
 619         if (required_type2 != required_type) {
 620             errmsg += " or ";
 621             errmsg += str(int(required_type2));
 622         }
 623         errmsg += ", got ";
 624         errmsg += str(type);
 625         throw Xapian::NetworkError(errmsg);
 626     }
 627
 628     return static_cast<reply_type>(type);
 629 }
 630
 631 void
 632 RemoteDatabase::send_message(message_type type, string_view message) const
 633 {
 634     double end_time = RealTime::end_time(timeout);
 635     while (pending_reply) {
 636         string dummy;
 637         int reply_code = link.get_message(dummy, end_time);
 638         if (reply_code < 0)
 639             throw_connection_closed_unexpectedly();
 640         if (!is_intermediate_reply(reply_code)) {
 641             pending_reply = false;
 642         }
 643     }
 644     link.send_message(static_cast<unsigned char>(type), message, end_time);
 645     pending_reply = true;
 646 }
 647
 648 void
 649 RemoteDatabase::do_close()
 650 {
 651     if (!is_read_only()) {
 652         try {
 653             if (transaction_active()) {
 654                 end_transaction(false);
 655             } else {
 656                 commit();
 657             }
 658         } catch (...) {
 659             try {
 660                 link.do_close();
 661             } catch (...) {
 662             }
 663             throw;
 664         }
 665
 666         // If we're writable, send a shutdown message to the server and wait
 667         // for it to close its end of the connection so we know that changes
 668         // have been written and flushed, and the database write lock released.
 669         // For the non-writable case, there's no need to wait - it would just
 670         // slow down searching needlessly.
 671         link.shutdown();
 672     }
 673     link.do_close();
 674 }
 675
 676 void
 677 RemoteDatabase::set_query(const Xapian::Query& query,
 678                           Xapian::termcount qlen,
 679                           Xapian::valueno collapse_key,
 680                           Xapian::doccount collapse_max,
 681                           Xapian::Enquire::docid_order order,
 682                           Xapian::valueno sort_key,
 683                           Xapian::Enquire::Internal::sort_setting sort_by,
 684                           bool sort_value_forward,
 685                           double time_limit,
 686                           int percent_threshold, double weight_threshold,
 687                           const Xapian::Weight& wtscheme,
 688                           const Xapian::RSet &omrset,
 689                           const vector<opt_ptr_spy>& matchspies) const
 690 {
 691     string message;
 692     pack_string(message, query.serialise());
 693
 694     // Serialise assorted Enquire settings.
 695     pack_uint(message, qlen);
 696     pack_uint(message, collapse_max);
 697     if (collapse_max) pack_uint(message, collapse_key);
 698     message += char(order);
 699     message += char(sort_by);
 700     if (sort_by != Xapian::Enquire::Internal::REL) {
 701         pack_uint(message, sort_key);
 702     }
 703     pack_bool(message, sort_value_forward);
 704     message += serialise_double(time_limit);
 705     message += char(percent_threshold);
 706     message += serialise_double(weight_threshold);
 707
 708     pack_string(message, wtscheme.name());
 709
 710     pack_string(message, wtscheme.serialise());
 711
 712     pack_string(message, serialise_rset(omrset));
 713
 714     for (auto i : matchspies) {
 715         const string& name = i->name();
 716         if (name.empty()) {
 717             throw Xapian::UnimplementedError("MatchSpy subclass not suitable for use with remote searches - name() method returned empty string");
 718         }
 719         pack_string(message, name);
 720         pack_string(message, i->serialise());
 721     }
 722
 723     send_message(MSG_QUERY, message);
 724 }
 725
 726 void
 727 RemoteDatabase::accumulate_remote_stats(Xapian::Weight::Internal& total) const
 728 {
 729     string message;
 730     get_message(message, REPLY_STATS);
 731     const char* p = message.data();
 732     Xapian::Weight::Internal remote_stats;
 733     unserialise_stats(p, p + message.size(), remote_stats);
 734     total += remote_stats;
 735 }
 736
 737 void
 738 RemoteDatabase::send_global_stats(Xapian::doccount first,
 739                                   Xapian::doccount maxitems,
 740                                   Xapian::doccount check_at_least,
 741                                   const Xapian::KeyMaker* sorter,
 742                                   const Xapian::Weight::Internal &stats) const
 743 {
 744     string message;
 745     pack_uint(message, first);
 746     pack_uint(message, maxitems);
 747     pack_uint(message, check_at_least);
 748     if (!sorter) {
 749         pack_string_empty(message);
 750     } else {
 751         const string& name = sorter->name();
 752         if (name.empty()) {
 753             throw_invalid_operation("sorter reported empty name");
 754         }
 755         pack_string(message, name);
 756         pack_string(message, sorter->serialise());
 757     }
 758     message += serialise_stats(stats);
 759     send_message(MSG_GETMSET, message);
 760 }
 761
 762 Xapian::MSet
 763 RemoteDatabase::get_mset(const vector<opt_ptr_spy>& matchspies) const
 764 {
 765     string message;
 766     get_message(message, REPLY_RESULTS);
 767     const char * p = message.data();
 768     const char * p_end = p + message.size();
 769
 770     string spyresults;
 771     for (auto i : matchspies) {
 772         if (!unpack_string(&p, p_end, spyresults)) {
 773             throw Xapian::NetworkError("Expected serialised matchspy");
 774         }
 775         i->merge_results(spyresults);
 776     }
 777     Xapian::MSet mset;
 778     mset.internal->unserialise(p, p_end);
 779     return mset;
 780 }
 781
 782 void
 783 RemoteDatabase::commit()
 784 {
 785     if (!uncommitted_changes) return;
 786
 787     send_message(MSG_COMMIT, {});
 788
 789     // We need to wait for a response to ensure documents have been committed.
 790     string message;
 791     get_message(message, REPLY_DONE);
 792
 793     uncommitted_changes = false;
 794 }
 795
 796 void
 797 RemoteDatabase::cancel()
 798 {
 799     if (!uncommitted_changes) return;
 800
 801     cached_stats_valid = false;
 802     mru_slot = Xapian::BAD_VALUENO;
 803
 804     send_message(MSG_CANCEL, {});
 805     string dummy;
 806     get_message(dummy, REPLY_DONE);
 807
 808     uncommitted_changes = false;
 809 }
 810
 811 Xapian::docid
 812 RemoteDatabase::add_document(const Xapian::Document & doc)
 813 {
 814     cached_stats_valid = false;
 815     mru_slot = Xapian::BAD_VALUENO;
 816     uncommitted_changes = true;
 817
 818     send_message(MSG_ADDDOCUMENT, serialise_document(doc));
 819
 820     string message;
 821     get_message(message, REPLY_ADDDOCUMENT);
 822
 823     const char* p = message.data();
 824     const char* p_end = p + message.size();
 825     Xapian::docid did;
 826     if (!unpack_uint_last(&p, p_end, &did)) {
 827         unpack_throw_serialisation_error(p);
 828     }
 829     return did;
 830 }
 831
 832 void
 833 RemoteDatabase::delete_document(Xapian::docid did)
 834 {
 835     cached_stats_valid = false;
 836     mru_slot = Xapian::BAD_VALUENO;
 837     uncommitted_changes = true;
 838
 839     string message;
 840     pack_uint_last(message, did);
 841     send_message(MSG_DELETEDOCUMENT, message);
 842
 843     get_message(message, REPLY_DONE);
 844 }
 845
 846 void
 847 RemoteDatabase::delete_document(std::string_view unique_term)
 848 {
 849     cached_stats_valid = false;
 850     mru_slot = Xapian::BAD_VALUENO;
 851     uncommitted_changes = true;
 852
 853     send_message(MSG_DELETEDOCUMENTTERM, unique_term);
 854     string dummy;
 855     get_message(dummy, REPLY_DONE);
 856 }
 857
 858 void
 859 RemoteDatabase::replace_document(Xapian::docid did,
 860                                  const Xapian::Document & doc)
 861 {
 862     cached_stats_valid = false;
 863     mru_slot = Xapian::BAD_VALUENO;
 864     uncommitted_changes = true;
 865
 866     string message;
 867     pack_uint(message, did);
 868     message += serialise_document(doc);
 869
 870     send_message(MSG_REPLACEDOCUMENT, message);
 871
 872     get_message(message, REPLY_DONE);
 873 }
 874
 875 Xapian::docid
 876 RemoteDatabase::replace_document(std::string_view unique_term,
 877                                  const Xapian::Document & doc)
 878 {
 879     cached_stats_valid = false;
 880     mru_slot = Xapian::BAD_VALUENO;
 881     uncommitted_changes = true;
 882
 883     string message;
 884     pack_string(message, unique_term);
 885     message += serialise_document(doc);
 886
 887     send_message(MSG_REPLACEDOCUMENTTERM, message);
 888
 889     get_message(message, REPLY_ADDDOCUMENT);
 890
 891     const char* p = message.data();
 892     const char* p_end = p + message.size();
 893     Xapian::docid did;
 894     if (!unpack_uint_last(&p, p_end, &did)) {
 895         unpack_throw_serialisation_error(p);
 896     }
 897     return did;
 898 }
 899
 900 string
 901 RemoteDatabase::get_uuid() const
 902 {
 903     return uuid;
 904 }
 905
 906 string
 907 RemoteDatabase::get_metadata(string_view key) const
 908 {
 909     send_message(MSG_GETMETADATA, key);
 910     string metadata;
 911     get_message(metadata, REPLY_METADATA);
 912     return metadata;
 913 }
 914
 915 void
 916 RemoteDatabase::set_metadata(string_view key, string_view value)
 917 {
 918     uncommitted_changes = true;
 919
 920     string message;
 921     pack_string(message, key);
 922     message += value;
 923     send_message(MSG_SETMETADATA, message);
 924
 925     get_message(message, REPLY_DONE);
 926 }
 927
 928 void
 929 RemoteDatabase::request_document(Xapian::docid did) const
 930 {
 931     string message;
 932     pack_uint(message, did);
 933     send_message(MSG_REQUESTDOCUMENT, message);
 934
 935     get_message(message, REPLY_DONE);
 936 }
 937
 938 void
 939 RemoteDatabase::add_spelling(string_view word,
 940                              Xapian::termcount freqinc) const
 941 {
 942     uncommitted_changes = true;
 943
 944     string message;
 945     pack_uint(message, freqinc);
 946     message += word;
 947     send_message(MSG_ADDSPELLING, message);
 948
 949     get_message(message, REPLY_DONE);
 950 }
 951
 952 Xapian::termcount
 953 RemoteDatabase::remove_spelling(string_view word,
 954                                 Xapian::termcount freqdec) const
 955 {
 956     uncommitted_changes = true;
 957
 958     string message;
 959     pack_uint(message, freqdec);
 960     message += word;
 961     send_message(MSG_REMOVESPELLING, message);
 962
 963     get_message(message, REPLY_REMOVESPELLING);
 964     const char * p = message.data();
 965     const char * p_end = p + message.size();
 966     Xapian::termcount result;
 967     if (!unpack_uint_last(&p, p_end, &result)) {
 968         throw Xapian::NetworkError("Bad REPLY_REMOVESPELLING",
 969                                    link.get_context());
 970     }
 971     return result;
 972 }
 973
 974 TermList*
 975 RemoteDatabase::open_synonym_termlist(string_view word) const
 976 {
 977     string message;
 978     send_message(MSG_SYNONYMTERMLIST, word);
 979     get_message(message, REPLY_SYNONYMTERMLIST);
 980     return new RemoteKeyList({}, std::move(message));
 981 }
 982
 983 TermList*
 984 RemoteDatabase::open_synonym_keylist(string_view prefix) const
 985 {
 986     string message;
 987     send_message(MSG_SYNONYMKEYLIST, prefix);
 988     get_message(message, REPLY_SYNONYMKEYLIST);
 989     return new RemoteKeyList({}, std::move(message));
 990 }
 991
 992 void
 993 RemoteDatabase::add_synonym(string_view word, string_view synonym) const
 994 {
 995     uncommitted_changes = true;
 996
 997     string message;
 998     pack_string(message, word);
 999     message += synonym;
1000     send_message(MSG_ADDSYNONYM, message);
1001     get_message(message, REPLY_DONE);
1002 }
1003
1004 void
1005 RemoteDatabase::remove_synonym(string_view word, string_view synonym) const
1006 {
1007     uncommitted_changes = true;
1008
1009     string message;
1010     pack_string(message, word);
1011     message += synonym;
1012     send_message(MSG_REMOVESYNONYM, message);
1013     get_message(message, REPLY_DONE);
1014 }
1015
1016 void
1017 RemoteDatabase::clear_synonyms(string_view word) const
1018 {
1019     uncommitted_changes = true;
1020
1021     string message;
1022     send_message(MSG_CLEARSYNONYMS, word);
1023     get_message(message, REPLY_DONE);
1024 }
1025
1026 bool
1027 RemoteDatabase::locked() const
1028 {
1029     throw Xapian::UnimplementedError("Database::locked() not implemented for remote backend");
1030 }
1031
1032 string
1033 RemoteDatabase::reconstruct_text(Xapian::docid did,
1034                                  size_t length,
1035                                  string_view prefix,
1036                                  Xapian::termpos start_pos,
1037                                  Xapian::termpos end_pos) const
1038 {
1039     string message;
1040     pack_uint(message, did);
1041     pack_uint(message, length);
1042     pack_uint(message, start_pos);
1043     pack_uint(message, end_pos);
1044     message += prefix;
1045     send_message(MSG_RECONSTRUCTTEXT, message);
1046
1047     get_message(message, REPLY_RECONSTRUCTTEXT);
1048     return message;
1049 }
1050
1051 string
1052 RemoteDatabase::get_description() const
1053 {
1054     string desc = "Remote(context=";
1055     desc += link.get_context();
1056     desc += ')';
1057     return desc;
1058 }