xapian-core/net/remoteserver.cc

   1 /** @file
   2  *  @brief Xapian remote backend server base class
   3  */
   4 /* Copyright (C) 2006-2024 Olly Betts
   5  * Copyright (C) 2006,2007,2009,2010 Lemur Consulting Ltd
   6  *
   7  * This program is free software; you can redistribute it and/or modify
   8  * it under the terms of the GNU General Public License as published by
   9  * the Free Software Foundation; either version 2 of the License, or
  10  * (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 #include "remoteserver.h"
  24
  25 #include "xapian/constants.h"
  26 #include "xapian/database.h"
  27 #include "xapian/enquire.h"
  28 #include "xapian/error.h"
  29 #include "xapian/matchspy.h"
  30 #include "xapian/query.h"
  31 #include "xapian/rset.h"
  32 #include "xapian/valueiterator.h"
  33
  34 #include <signal.h>
  35 #include <cerrno>
  36 #include <cstdlib>
  37 #include <memory>
  38
  39 #include "api/msetinternal.h"
  40 #include "api/termlist.h"
  41 #include "matcher/matcher.h"
  42 #include "omassert.h"
  43 #include "pack.h"
  44 #include "realtime.h"
  45 #include "serialise.h"
  46 #include "serialise-double.h"
  47 #include "serialise-error.h"
  48 #include "str.h"
  49 #include "stringutils.h"
  50 #include "weight/weightinternal.h"
  51
  52 using namespace std;
  53
  54 [[noreturn]]
  55 static void
  56 throw_read_only()
  57 {
  58     throw Xapian::InvalidOperationError("Server is read-only");
  59 }
  60
  61 /// Class to throw when we receive the connection closing message.
  62 struct ConnectionClosed { };
  63
  64 RemoteServer::RemoteServer(const vector<string>& dbpaths,
  65                            int fdin_, int fdout_,
  66                            double active_timeout_, double idle_timeout_,
  67                            bool writable_)
  68     : RemoteConnection(fdin_, fdout_, string()),
  69       writable(writable_),
  70       active_timeout(active_timeout_), idle_timeout(idle_timeout_)
  71 {
  72     // Catch errors opening the database and propagate them to the client.
  73     try {
  74         Assert(!dbpaths.empty());
  75         // We always open the database read-only to start with.  If we're
  76         // writable, the client can ask to be upgraded to write access once
  77         // connected if it wants it.
  78         db = new Xapian::Database(dbpaths[0]);
  79         // Build a better description than Database::get_description() gives
  80         // in the variable context.  FIXME: improve Database::get_description()
  81         // and then just use that instead.
  82         context = dbpaths[0];
  83
  84         vector<string>::const_iterator i(dbpaths.begin());
  85         for (++i; i != dbpaths.end(); ++i) {
  86             db->add_database(Xapian::Database(*i));
  87             context += ' ';
  88             context += *i;
  89         }
  90     } catch (const Xapian::Error &err) {
  91         // Propagate the exception to the client.
  92         send_message(REPLY_EXCEPTION, serialise_error(err));
  93         // And rethrow it so our caller can log it and close the connection.
  94         throw;
  95     }
  96
  97 #ifndef __WIN32__
  98     // It's simplest to just ignore SIGPIPE.  We'll still know if the
  99     // connection dies because we'll get EPIPE back from write().
 100     //
 101     // This is OK because RemoteServer subclasses are only used in
 102     // specialised programs - if we expose any of them as API classes
 103     // then we should use SO_NOSIGPIPE/MSG_NOSIGNAL instead like we do
 104     // on the client side.
 105     if (signal(SIGPIPE, SIG_IGN) == SIG_ERR)
 106         throw Xapian::NetworkError("Couldn't set SIGPIPE to SIG_IGN", errno);
 107 #endif
 108
 109     // Send greeting message.
 110     msg_update(string());
 111 }
 112
 113 RemoteServer::~RemoteServer()
 114 {
 115     delete db;
 116     // wdb is either NULL or equal to db, so we shouldn't delete it too!
 117 }
 118
 119 message_type
 120 RemoteServer::get_message(double timeout, string & result,
 121                           message_type required_type)
 122 {
 123     double end_time = RealTime::end_time(timeout);
 124     int type = RemoteConnection::get_message(result, end_time);
 125
 126     // Handle "shutdown connection" message here.  Treat EOF here for a read-only
 127     // database the same way since a read-only client just closes the
 128     // connection when done.
 129     if (type == MSG_SHUTDOWN || (type < 0 && wdb == NULL))
 130         throw ConnectionClosed();
 131     if (type < 0)
 132         throw Xapian::NetworkError("Connection closed unexpectedly");
 133     if (type >= MSG_MAX) {
 134         string errmsg("Invalid message type ");
 135         errmsg += str(type);
 136         throw Xapian::NetworkError(errmsg);
 137     }
 138     if (required_type != MSG_MAX && type != int(required_type)) {
 139         string errmsg("Expecting message type ");
 140         errmsg += str(int(required_type));
 141         errmsg += ", got ";
 142         errmsg += str(type);
 143         throw Xapian::NetworkError(errmsg);
 144     }
 145     return static_cast<message_type>(type);
 146 }
 147
 148 void
 149 RemoteServer::send_message(reply_type type, string_view message)
 150 {
 151     double end_time = RealTime::end_time(active_timeout);
 152     unsigned char type_as_char = static_cast<unsigned char>(type);
 153     RemoteConnection::send_message(type_as_char, message, end_time);
 154 }
 155
 156 typedef void (RemoteServer::* dispatch_func)(const string &);
 157
 158 void
 159 RemoteServer::run()
 160 {
 161     while (true) {
 162         try {
 163             string message;
 164             size_t type = get_message(idle_timeout, message);
 165             switch (type) {
 166                 case MSG_ALLTERMS:
 167                     msg_allterms(message);
 168                     continue;
 169                 case MSG_COLLFREQ:
 170                     msg_collfreq(message);
 171                     continue;
 172                 case MSG_DOCUMENT:
 173                     msg_document(message);
 174                     continue;
 175                 case MSG_TERMEXISTS:
 176                     msg_termexists(message);
 177                     continue;
 178                 case MSG_TERMFREQ:
 179                     msg_termfreq(message);
 180                     continue;
 181                 case MSG_VALUESTATS:
 182                     msg_valuestats(message);
 183                     continue;
 184                 case MSG_KEEPALIVE:
 185                     msg_keepalive(message);
 186                     continue;
 187                 case MSG_DOCLENGTH:
 188                     msg_doclength(message);
 189                     continue;
 190                 case MSG_QUERY:
 191                     msg_query(message);
 192                     continue;
 193                 case MSG_TERMLIST:
 194                     msg_termlist(message);
 195                     continue;
 196                 case MSG_POSITIONLIST:
 197                     msg_positionlist(message);
 198                     continue;
 199                 case MSG_POSTLIST:
 200                     msg_postlist(message);
 201                     continue;
 202                 case MSG_REOPEN:
 203                     msg_reopen(message);
 204                     continue;
 205                 case MSG_UPDATE:
 206                     msg_update(message);
 207                     continue;
 208                 case MSG_ADDDOCUMENT:
 209                     msg_adddocument(message);
 210                     continue;
 211                 case MSG_CANCEL:
 212                     msg_cancel(message);
 213                     continue;
 214                 case MSG_DELETEDOCUMENTTERM:
 215                     msg_deletedocumentterm(message);
 216                     continue;
 217                 case MSG_COMMIT:
 218                     msg_commit(message);
 219                     continue;
 220                 case MSG_REPLACEDOCUMENT:
 221                     msg_replacedocument(message);
 222                     continue;
 223                 case MSG_REPLACEDOCUMENTTERM:
 224                     msg_replacedocumentterm(message);
 225                     continue;
 226                 case MSG_DELETEDOCUMENT:
 227                     msg_deletedocument(message);
 228                     continue;
 229                 case MSG_WRITEACCESS:
 230                     msg_writeaccess(message);
 231                     continue;
 232                 case MSG_GETMETADATA:
 233                     msg_getmetadata(message);
 234                     continue;
 235                 case MSG_SETMETADATA:
 236                     msg_setmetadata(message);
 237                     continue;
 238                 case MSG_REQUESTDOCUMENT:
 239                     msg_requestdocument(message);
 240                     continue;
 241                 case MSG_ADDSPELLING:
 242                     msg_addspelling(message);
 243                     continue;
 244                 case MSG_REMOVESPELLING:
 245                     msg_removespelling(message);
 246                     continue;
 247                 case MSG_METADATAKEYLIST:
 248                     msg_metadatakeylist(message);
 249                     continue;
 250                 case MSG_FREQS:
 251                     msg_freqs(message);
 252                     continue;
 253                 case MSG_UNIQUETERMS:
 254                     msg_uniqueterms(message);
 255                     continue;
 256                 case MSG_WDFDOCMAX:
 257                     msg_wdfdocmax(message);
 258                     continue;
 259                 case MSG_POSITIONLISTCOUNT:
 260                     msg_positionlistcount(message);
 261                     continue;
 262                 case MSG_RECONSTRUCTTEXT:
 263                     msg_reconstructtext(message);
 264                     continue;
 265                 case MSG_SYNONYMTERMLIST:
 266                     msg_synonymtermlist(message);
 267                     continue;
 268                 case MSG_SYNONYMKEYLIST:
 269                     msg_synonymkeylist(message);
 270                     continue;
 271                 case MSG_ADDSYNONYM:
 272                     msg_addsynonym(message);
 273                     continue;
 274                 case MSG_REMOVESYNONYM:
 275                     msg_removesynonym(message);
 276                     continue;
 277                 case MSG_CLEARSYNONYMS:
 278                     msg_clearsynonyms(message);
 279                     continue;
 280                 default: {
 281                     // MSG_GETMSET - used during a conversation.
 282                     // MSG_SHUTDOWN - handled by get_message().
 283                     string errmsg("Unexpected message type ");
 284                     errmsg += str(type);
 285                     throw Xapian::InvalidArgumentError(errmsg);
 286                 }
 287             }
 288         } catch (const Xapian::NetworkTimeoutError & e) {
 289             try {
 290                 // We've had a timeout, so the client may not be listening, so
 291                 // set the end_time to 1 and if we can't send the message right
 292                 // away, just exit and the client will cope.
 293                 send_message(REPLY_EXCEPTION, serialise_error(e), 1.0);
 294             } catch (...) {
 295             }
 296             // And rethrow it so our caller can log it and close the
 297             // connection.
 298             throw;
 299         } catch (const Xapian::NetworkError &) {
 300             // All other network errors mean we are fatally confused and are
 301             // unlikely to be able to communicate further across this
 302             // connection.  So we don't try to propagate the error to the
 303             // client, but instead just rethrow the exception so our caller can
 304             // log it and close the connection.
 305             throw;
 306         } catch (const Xapian::Error &e) {
 307             // Propagate the exception to the client, then return to the main
 308             // message handling loop.
 309             send_message(REPLY_EXCEPTION, serialise_error(e));
 310         } catch (ConnectionClosed &) {
 311             return;
 312         } catch (...) {
 313             // Propagate an unknown exception to the client.
 314             send_message(REPLY_EXCEPTION, {});
 315             // And rethrow it so our caller can log it and close the
 316             // connection.
 317             throw;
 318         }
 319     }
 320 }
 321
 322 void
 323 RemoteServer::msg_allterms(const string& message)
 324 {
 325     string reply;
 326     string prev = message;
 327     const string& prefix = message;
 328     for (Xapian::TermIterator t = db->allterms_begin(prefix);
 329          t != db->allterms_end(prefix);
 330          ++t) {
 331         const string& term = *t;
 332         size_t reuse = common_prefix_length(prev, term, 255);
 333         reply.append(1, char(reuse));
 334         pack_uint(reply, term.size() - reuse);
 335         reply.append(term, reuse, string::npos);
 336         pack_uint(reply, t.get_termfreq());
 337         prev = term;
 338     }
 339     send_message(REPLY_ALLTERMS, reply);
 340 }
 341
 342 void
 343 RemoteServer::msg_termlist(const string &message)
 344 {
 345     const char *p = message.data();
 346     const char *p_end = p + message.size();
 347     Xapian::docid did;
 348     if (!unpack_uint_last(&p, p_end, &did)) {
 349         throw Xapian::NetworkError("Bad MSG_TERMLIST");
 350     }
 351     Xapian::TermIterator t = db->termlist_begin(did);
 352     Xapian::termcount num_terms = 0;
 353     if (t.internal)
 354         num_terms = t.internal->get_approx_size();
 355     string reply;
 356     pack_uint(reply, db->get_doclength(did));
 357     pack_uint_last(reply, num_terms);
 358     send_message(REPLY_TERMLISTHEADER, reply);
 359
 360     reply.resize(0);
 361     string prev;
 362     while (t != db->termlist_end(did)) {
 363         const string& term = *t;
 364         size_t reuse = common_prefix_length(prev, term, 255);
 365         reply.append(1, char(reuse));
 366         pack_uint(reply, term.size() - reuse);
 367         reply.append(term, reuse, string::npos);
 368         pack_uint(reply, t.get_wdf());
 369         pack_uint(reply, t.get_termfreq());
 370         prev = term;
 371         ++t;
 372     }
 373     send_message(REPLY_TERMLIST, reply);
 374 }
 375
 376 void
 377 RemoteServer::msg_positionlist(const string &message)
 378 {
 379     const char *p = message.data();
 380     const char *p_end = p + message.size();
 381     Xapian::docid did;
 382     if (!unpack_uint(&p, p_end, &did)) {
 383         throw Xapian::NetworkError("Bad MSG_POSITIONLIST");
 384     }
 385     string term(p, p_end - p);
 386
 387     string reply;
 388     Xapian::termpos lastpos = static_cast<Xapian::termpos>(-1);
 389     for (Xapian::PositionIterator i = db->positionlist_begin(did, term);
 390          i != db->positionlist_end(did, term);
 391          ++i) {
 392         Xapian::termpos pos = *i;
 393         pack_uint(reply, UNSIGNED_OVERFLOW_OK(pos - lastpos - 1));
 394         lastpos = pos;
 395     }
 396     send_message(REPLY_POSITIONLIST, reply);
 397 }
 398
 399 void
 400 RemoteServer::msg_positionlistcount(const string &message)
 401 {
 402     const char *p = message.data();
 403     const char *p_end = p + message.size();
 404     Xapian::docid did;
 405     if (!unpack_uint(&p, p_end, &did)) {
 406         throw Xapian::NetworkError("Bad MSG_POSITIONLISTCOUNT");
 407     }
 408
 409     // This is kind of clumsy, but what the public API requires.
 410     Xapian::termcount result = 0;
 411     Xapian::TermIterator termit = db->termlist_begin(did);
 412     if (termit != db->termlist_end(did)) {
 413         string term(p, p_end - p);
 414         termit.skip_to(term);
 415         if (termit != db->termlist_end(did) && *termit == term) {
 416             result = termit.positionlist_count();
 417         }
 418     }
 419     string reply;
 420     pack_uint_last(reply, result);
 421     send_message(REPLY_POSITIONLISTCOUNT, reply);
 422 }
 423
 424 void
 425 RemoteServer::msg_postlist(const string &message)
 426 {
 427     const string & term = message;
 428
 429     Xapian::doccount termfreq = db->get_termfreq(term);
 430     string reply;
 431     pack_uint_last(reply, termfreq);
 432     send_message(REPLY_POSTLISTHEADER, reply);
 433
 434     reply.resize(0);
 435     Xapian::docid lastdocid = 0;
 436     for (Xapian::PostingIterator i = db->postlist_begin(term);
 437          i != db->postlist_end(term);
 438          ++i) {
 439         Xapian::docid newdocid = *i;
 440         pack_uint(reply, newdocid - lastdocid - 1);
 441         pack_uint(reply, i.get_wdf());
 442
 443         lastdocid = newdocid;
 444     }
 445
 446     send_message(REPLY_POSTLIST, reply);
 447 }
 448
 449 void
 450 RemoteServer::msg_writeaccess(const string & msg)
 451 {
 452     if (!writable)
 453         throw_read_only();
 454
 455     int flags = 0;
 456     const char *p = msg.c_str();
 457     const char *p_end = p + msg.size();
 458     if (p != p_end) {
 459         unsigned flag_bits;
 460         if (!unpack_uint_last(&p, p_end, &flag_bits)) {
 461             throw Xapian::NetworkError("Bad flags in MSG_WRITEACCESS");
 462         }
 463         flags = flag_bits &~ Xapian::DB_ACTION_MASK_;
 464     }
 465
 466     wdb = new Xapian::WritableDatabase(db->lock(flags));
 467     delete db;
 468     db = wdb;
 469     msg_update(msg);
 470 }
 471
 472 void
 473 RemoteServer::msg_reopen(const string & msg)
 474 {
 475     if (!db->reopen()) {
 476         send_message(REPLY_DONE, {});
 477         return;
 478     }
 479     msg_update(msg);
 480 }
 481
 482 void
 483 RemoteServer::msg_update(const string &)
 484 {
 485     static const char protocol[2] = {
 486         char(XAPIAN_REMOTE_PROTOCOL_MAJOR_VERSION),
 487         char(XAPIAN_REMOTE_PROTOCOL_MINOR_VERSION)
 488     };
 489     string message(protocol, 2);
 490     Xapian::doccount num_docs = db->get_doccount();
 491     pack_uint(message, num_docs);
 492     pack_uint(message, db->get_lastdocid() - num_docs);
 493     Xapian::termcount doclen_lb = db->get_doclength_lower_bound();
 494     pack_uint(message, doclen_lb);
 495     pack_uint(message, db->get_doclength_upper_bound() - doclen_lb);
 496     pack_bool(message, db->has_positions());
 497     pack_uint(message, db->get_total_length());
 498     message += db->get_uuid();
 499     send_message(REPLY_UPDATE, message);
 500 }
 501
 502 void
 503 RemoteServer::msg_query(const string &message_in)
 504 {
 505     const char *p = message_in.c_str();
 506     const char *p_end = p + message_in.size();
 507
 508     // Unserialise the Query.
 509     string serialisation;
 510     if (!unpack_string(&p, p_end, serialisation)) {
 511         throw Xapian::NetworkError("Bad MSG_QUERY");
 512     }
 513
 514     Xapian::Query query(Xapian::Query::unserialise(serialisation, reg));
 515
 516     // Unserialise assorted Enquire settings.
 517     Xapian::termcount qlen;
 518     Xapian::valueno collapse_max;
 519     if (!unpack_uint(&p, p_end, &qlen) ||
 520         !unpack_uint(&p, p_end, &collapse_max)) {
 521         throw Xapian::NetworkError("Bad MSG_QUERY");
 522     }
 523
 524     Xapian::valueno collapse_key = Xapian::BAD_VALUENO;
 525     if (collapse_max) {
 526         if (!unpack_uint(&p, p_end, &collapse_key)) {
 527             throw Xapian::NetworkError("Bad MSG_QUERY");
 528         }
 529     }
 530
 531     if (p_end - p < 4 || static_cast<unsigned char>(*p) > 2) {
 532         throw Xapian::NetworkError("bad message (docid_order)");
 533     }
 534     Xapian::Enquire::docid_order order;
 535     order = static_cast<Xapian::Enquire::docid_order>(*p++);
 536
 537     if (static_cast<unsigned char>(*p) > 3) {
 538         throw Xapian::NetworkError("bad message (sort_by)");
 539     }
 540     Xapian::Enquire::Internal::sort_setting sort_by;
 541     sort_by = static_cast<Xapian::Enquire::Internal::sort_setting>(*p++);
 542
 543     Xapian::valueno sort_key = Xapian::BAD_VALUENO;
 544     if (sort_by != Xapian::Enquire::Internal::REL) {
 545         if (!unpack_uint(&p, p_end, &sort_key)) {
 546             throw Xapian::NetworkError("Bad MSG_QUERY");
 547         }
 548     }
 549
 550     bool sort_value_forward;
 551     if (!unpack_bool(&p, p_end, &sort_value_forward)) {
 552         throw Xapian::NetworkError("bad message (sort_value_forward)");
 553     }
 554
 555     double time_limit = unserialise_double(&p, p_end);
 556
 557     int percent_threshold = *p++;
 558     if (percent_threshold < 0 || percent_threshold > 100) {
 559         throw Xapian::NetworkError("bad message (percent_threshold)");
 560     }
 561
 562     double weight_threshold = unserialise_double(&p, p_end);
 563     if (weight_threshold < 0) {
 564         throw Xapian::NetworkError("bad message (weight_threshold)");
 565     }
 566
 567     // Unserialise the Weight object.
 568     string wtname;
 569     if (!unpack_string(&p, p_end, wtname)) {
 570         throw Xapian::NetworkError("Bad MSG_QUERY");
 571     }
 572
 573     const Xapian::Weight * wttype = reg.get_weighting_scheme(wtname);
 574     if (wttype == NULL) {
 575         // Note: user weighting schemes should be registered by adding them to
 576         // a Registry, and setting the context using
 577         // RemoteServer::set_registry().
 578         throw Xapian::InvalidArgumentError("Weighting scheme " +
 579                                            wtname + " not registered");
 580     }
 581
 582     if (!unpack_string(&p, p_end, serialisation)) {
 583         throw Xapian::NetworkError("Bad MSG_QUERY");
 584     }
 585     unique_ptr<Xapian::Weight> wt(wttype->unserialise(serialisation));
 586
 587     // Unserialise the RSet object.
 588     if (!unpack_string(&p, p_end, serialisation)) {
 589         throw Xapian::NetworkError("Bad MSG_QUERY");
 590     }
 591     Xapian::RSet rset = unserialise_rset(serialisation);
 592
 593     // Unserialise any MatchSpy objects.
 594     vector<Xapian::Internal::opt_intrusive_ptr<Xapian::MatchSpy>> matchspies;
 595     while (p != p_end) {
 596         string spytype;
 597         if (!unpack_string(&p, p_end, spytype)) {
 598             throw Xapian::NetworkError("Bad MSG_QUERY");
 599         }
 600         const Xapian::MatchSpy * spyclass = reg.get_match_spy(spytype);
 601         if (spyclass == NULL) {
 602             throw Xapian::InvalidArgumentError("Match spy " + spytype +
 603                                                " not registered");
 604         }
 605
 606         if (!unpack_string(&p, p_end, serialisation)) {
 607             throw Xapian::NetworkError("Bad MSG_QUERY");
 608         }
 609         matchspies.push_back(spyclass->unserialise(serialisation,
 610                                                    reg)->release());
 611     }
 612
 613     Xapian::Weight::Internal local_stats;
 614     Matcher matcher(*db,
 615                     query, qlen, &rset, local_stats, *wt,
 616                     false,
 617                     collapse_key, collapse_max,
 618                     percent_threshold, weight_threshold,
 619                     order, sort_key, sort_by, sort_value_forward, time_limit,
 620                     matchspies);
 621
 622     send_message(REPLY_STATS, serialise_stats(local_stats));
 623
 624     string message;
 625     get_message(active_timeout, message, MSG_GETMSET);
 626     p = message.c_str();
 627     p_end = p + message.size();
 628
 629     Xapian::termcount first;
 630     Xapian::termcount maxitems;
 631     Xapian::termcount check_at_least;
 632     string sorter_type;
 633     if (!unpack_uint(&p, p_end, &first) ||
 634         !unpack_uint(&p, p_end, &maxitems) ||
 635         !unpack_uint(&p, p_end, &check_at_least) ||
 636         !unpack_string(&p, p_end, sorter_type)) {
 637         throw Xapian::NetworkError("Bad MSG_GETMSET");
 638     }
 639     unique_ptr<Xapian::KeyMaker> sorter;
 640     if (!sorter_type.empty()) {
 641         const Xapian::KeyMaker* sorterclass = reg.get_key_maker(sorter_type);
 642         if (sorterclass == NULL) {
 643             throw Xapian::InvalidArgumentError("KeyMaker " + sorter_type +
 644                                                " not registered");
 645         }
 646
 647         string serialised_sorter;
 648         if (!unpack_string(&p, p_end, serialised_sorter)) {
 649             throw Xapian::NetworkError("Bad MSG_GETMSET");
 650         }
 651         sorter.reset(sorterclass->unserialise(serialised_sorter, reg));
 652     }
 653
 654     unique_ptr<Xapian::Weight::Internal> total_stats(new Xapian::Weight::Internal);
 655     unserialise_stats(p, p_end, *total_stats);
 656
 657     Xapian::MSet mset = matcher.get_mset(first, maxitems, check_at_least,
 658                                          *total_stats, *wt, 0, sorter.get(),
 659                                          collapse_key, collapse_max,
 660                                          percent_threshold, weight_threshold,
 661                                          order,
 662                                          sort_key, sort_by, sort_value_forward,
 663                                          time_limit, matchspies);
 664     // FIXME: The local side already has these stats, except for the maxpart
 665     // information.
 666     mset.internal->set_stats(total_stats.release());
 667
 668     message.resize(0);
 669     for (auto i : matchspies) {
 670         pack_string(message, i->serialise_results());
 671     }
 672     message += mset.internal->serialise();
 673     send_message(REPLY_RESULTS, message);
 674 }
 675
 676 void
 677 RemoteServer::msg_document(const string &message)
 678 {
 679     const char *p = message.data();
 680     const char *p_end = p + message.size();
 681     Xapian::docid did;
 682     if (!unpack_uint_last(&p, p_end, &did)) {
 683         throw Xapian::NetworkError("Bad MSG_DOCUMENT");
 684     }
 685
 686     Xapian::Document doc = db->get_document(did);
 687
 688     send_message(REPLY_DOCDATA, doc.get_data());
 689
 690     Xapian::ValueIterator i;
 691     for (i = doc.values_begin(); i != doc.values_end(); ++i) {
 692         string item;
 693         pack_uint(item, i.get_valueno());
 694         item += *i;
 695         send_message(REPLY_VALUE, item);
 696     }
 697     send_message(REPLY_DONE, {});
 698 }
 699
 700 void
 701 RemoteServer::msg_keepalive(const string &)
 702 {
 703     // Ensure *our* database stays alive, as it may contain remote databases!
 704     db->keep_alive();
 705     send_message(REPLY_DONE, {});
 706 }
 707
 708 void
 709 RemoteServer::msg_termexists(const string &term)
 710 {
 711     send_message((db->term_exists(term) ? REPLY_TERMEXISTS
 712                                         : REPLY_TERMDOESNTEXIST), {});
 713 }
 714
 715 void
 716 RemoteServer::msg_collfreq(const string &term)
 717 {
 718     string reply;
 719     pack_uint_last(reply, db->get_collection_freq(term));
 720     send_message(REPLY_COLLFREQ, reply);
 721 }
 722
 723 void
 724 RemoteServer::msg_termfreq(const string &term)
 725 {
 726     string reply;
 727     pack_uint_last(reply, db->get_termfreq(term));
 728     send_message(REPLY_TERMFREQ, reply);
 729 }
 730
 731 void
 732 RemoteServer::msg_freqs(const string &term)
 733 {
 734     string msg;
 735     pack_uint(msg, db->get_termfreq(term));
 736     pack_uint_last(msg, db->get_collection_freq(term));
 737     send_message(REPLY_FREQS, msg);
 738 }
 739
 740 void
 741 RemoteServer::msg_valuestats(const string & message)
 742 {
 743     const char *p = message.data();
 744     const char *p_end = p + message.size();
 745     Xapian::valueno slot;
 746     if (!unpack_uint_last(&p, p_end, &slot)) {
 747         throw Xapian::NetworkError("Bad MSG_VALUESTATS");
 748     }
 749     string message_out;
 750     pack_uint(message_out, db->get_value_freq(slot));
 751     pack_string(message_out, db->get_value_lower_bound(slot));
 752     message_out += db->get_value_upper_bound(slot);
 753
 754     send_message(REPLY_VALUESTATS, message_out);
 755 }
 756
 757 void
 758 RemoteServer::msg_doclength(const string &message)
 759 {
 760     const char *p = message.data();
 761     const char *p_end = p + message.size();
 762     Xapian::docid did;
 763     if (!unpack_uint_last(&p, p_end, &did)) {
 764         throw Xapian::NetworkError("Bad MSG_DOCLENGTH");
 765     }
 766     string reply;
 767     pack_uint_last(reply, db->get_doclength(did));
 768     send_message(REPLY_DOCLENGTH, reply);
 769 }
 770
 771 void
 772 RemoteServer::msg_uniqueterms(const string &message)
 773 {
 774     const char *p = message.data();
 775     const char *p_end = p + message.size();
 776     Xapian::docid did;
 777     if (!unpack_uint_last(&p, p_end, &did)) {
 778         throw Xapian::NetworkError("Bad MSG_UNIQUETERMS");
 779     }
 780     string reply;
 781     pack_uint_last(reply, db->get_unique_terms(did));
 782     send_message(REPLY_UNIQUETERMS, reply);
 783 }
 784
 785 void
 786 RemoteServer::msg_wdfdocmax(const string& message)
 787 {
 788     const char* p = message.data();
 789     const char* p_end = p + message.size();
 790     Xapian::docid did;
 791     if (!unpack_uint_last(&p, p_end, &did)) {
 792         throw Xapian::NetworkError("Bad MSG_WDFDOCMAX");
 793     }
 794     string reply;
 795     pack_uint_last(reply, db->get_wdfdocmax(did));
 796     send_message(REPLY_WDFDOCMAX, reply);
 797 }
 798
 799 void
 800 RemoteServer::msg_reconstructtext(const string& message)
 801 {
 802     const char* p = message.data();
 803     const char* p_end = p + message.size();
 804     Xapian::docid did;
 805     size_t length;
 806     Xapian::termpos start_pos, end_pos;
 807     if (!unpack_uint(&p, p_end, &did) ||
 808         !unpack_uint(&p, p_end, &length) ||
 809         !unpack_uint(&p, p_end, &start_pos) ||
 810         !unpack_uint(&p, p_end, &end_pos)) {
 811         throw Xapian::NetworkError("Bad MSG_RECONSTRUCTTEXT");
 812     }
 813     send_message(REPLY_RECONSTRUCTTEXT,
 814                  db->reconstruct_text(did, length, {p, size_t(p_end - p)},
 815                                       start_pos, end_pos));
 816 }
 817
 818 void
 819 RemoteServer::msg_commit(const string &)
 820 {
 821     if (!wdb)
 822         throw_read_only();
 823
 824     wdb->commit();
 825
 826     send_message(REPLY_DONE, {});
 827 }
 828
 829 void
 830 RemoteServer::msg_cancel(const string &)
 831 {
 832     if (!wdb)
 833         throw_read_only();
 834
 835     // We can't call cancel since that's an internal method, but this
 836     // has the same effect with minimal additional overhead.
 837     wdb->begin_transaction(false);
 838     wdb->cancel_transaction();
 839
 840     send_message(REPLY_DONE, {});
 841 }
 842
 843 void
 844 RemoteServer::msg_adddocument(const string & message)
 845 {
 846     if (!wdb)
 847         throw_read_only();
 848
 849     Xapian::docid did = wdb->add_document(unserialise_document(message));
 850
 851     string reply;
 852     pack_uint_last(reply, did);
 853     send_message(REPLY_ADDDOCUMENT, reply);
 854 }
 855
 856 void
 857 RemoteServer::msg_deletedocument(const string & message)
 858 {
 859     if (!wdb)
 860         throw_read_only();
 861
 862     const char *p = message.data();
 863     const char *p_end = p + message.size();
 864     Xapian::docid did;
 865     if (!unpack_uint_last(&p, p_end, &did)) {
 866         throw Xapian::NetworkError("Bad MSG_DELETEDOCUMENT");
 867     }
 868
 869     wdb->delete_document(did);
 870
 871     send_message(REPLY_DONE, {});
 872 }
 873
 874 void
 875 RemoteServer::msg_deletedocumentterm(const string & message)
 876 {
 877     if (!wdb)
 878         throw_read_only();
 879
 880     wdb->delete_document(message);
 881
 882     send_message(REPLY_DONE, {});
 883 }
 884
 885 void
 886 RemoteServer::msg_replacedocument(const string & message)
 887 {
 888     if (!wdb)
 889         throw_read_only();
 890
 891     const char *p = message.data();
 892     const char *p_end = p + message.size();
 893     Xapian::docid did;
 894     if (!unpack_uint(&p, p_end, &did)) {
 895         throw Xapian::NetworkError("Bad MSG_REPLACEDOCUMENT");
 896     }
 897
 898     wdb->replace_document(did, unserialise_document(string(p, p_end)));
 899
 900     send_message(REPLY_DONE, {});
 901 }
 902
 903 void
 904 RemoteServer::msg_replacedocumentterm(const string & message)
 905 {
 906     if (!wdb)
 907         throw_read_only();
 908
 909     const char *p = message.data();
 910     const char *p_end = p + message.size();
 911     string unique_term;
 912     if (!unpack_string(&p, p_end, unique_term)) {
 913         throw Xapian::NetworkError("Bad MSG_REPLACEDOCUMENTTERM");
 914     }
 915     Xapian::docid did = wdb->replace_document(unique_term, unserialise_document(string(p, p_end)));
 916
 917     string reply;
 918     pack_uint_last(reply, did);
 919     send_message(REPLY_ADDDOCUMENT, reply);
 920 }
 921
 922 void
 923 RemoteServer::msg_getmetadata(const string & message)
 924 {
 925     send_message(REPLY_METADATA, db->get_metadata(message));
 926 }
 927
 928 void
 929 RemoteServer::msg_metadatakeylist(const string& message)
 930 {
 931     string reply;
 932     string prev = message;
 933     const string& prefix = message;
 934     for (Xapian::TermIterator t = db->metadata_keys_begin(prefix);
 935          t != db->metadata_keys_end(prefix);
 936          ++t) {
 937         const string& term = *t;
 938         size_t reuse = common_prefix_length(prev, term, 255);
 939         reply.append(1, char(reuse));
 940         pack_uint(reply, term.size() - reuse);
 941         reply.append(term, reuse, string::npos);
 942         prev = term;
 943     }
 944     send_message(REPLY_METADATAKEYLIST, reply);
 945 }
 946
 947 void
 948 RemoteServer::msg_setmetadata(const string & message)
 949 {
 950     if (!wdb)
 951         throw_read_only();
 952     const char *p = message.data();
 953     const char *p_end = p + message.size();
 954     string key;
 955     if (!unpack_string(&p, p_end, key)) {
 956         throw Xapian::NetworkError("Bad MSG_SETMETADATA");
 957     }
 958     string val(p, p_end - p);
 959     wdb->set_metadata(key, val);
 960
 961     send_message(REPLY_DONE, {});
 962 }
 963
 964 void
 965 RemoteServer::msg_requestdocument(const string& message)
 966 {
 967     const char* p = message.data();
 968     const char* p_end = p + message.size();
 969     Xapian::docid did;
 970     if (!unpack_uint_last(&p, p_end, &did)) {
 971         throw Xapian::NetworkError("Bad MSG_REQUESTDOCUMENT");
 972     }
 973     db->internal->request_document(did);
 974
 975     send_message(REPLY_DONE, string());
 976 }
 977
 978 void
 979 RemoteServer::msg_addspelling(const string & message)
 980 {
 981     if (!wdb)
 982         throw_read_only();
 983     const char *p = message.data();
 984     const char *p_end = p + message.size();
 985     Xapian::termcount freqinc;
 986     if (!unpack_uint(&p, p_end, &freqinc)) {
 987         throw Xapian::NetworkError("Bad MSG_ADDSPELLING");
 988     }
 989     wdb->add_spelling(string(p, p_end - p), freqinc);
 990
 991     send_message(REPLY_DONE, {});
 992 }
 993
 994 void
 995 RemoteServer::msg_removespelling(const string & message)
 996 {
 997     if (!wdb)
 998         throw_read_only();
 999     const char *p = message.data();
1000     const char *p_end = p + message.size();
1001     Xapian::termcount freqdec;
1002     if (!unpack_uint(&p, p_end, &freqdec)) {
1003         throw Xapian::NetworkError("Bad MSG_REMOVESPELLING");
1004     }
1005     string reply;
1006     pack_uint_last(reply, wdb->remove_spelling(string(p, p_end - p), freqdec));
1007     send_message(REPLY_REMOVESPELLING, reply);
1008 }
1009
1010 void
1011 RemoteServer::msg_synonymtermlist(const string& message)
1012 {
1013     Xapian::TermIterator t = db->synonyms_begin(message);
1014     string reply, prev;
1015     while (t != db->synonyms_end(message)) {
1016         const string& term = *t;
1017         size_t reuse = common_prefix_length(prev, term, 255);
1018         reply.append(1, char(reuse));
1019         pack_uint(reply, term.size() - reuse);
1020         reply.append(term, reuse, string::npos);
1021         prev = term;
1022         ++t;
1023     }
1024     send_message(REPLY_SYNONYMTERMLIST, reply);
1025 }
1026
1027 void
1028 RemoteServer::msg_synonymkeylist(const string& message)
1029 {
1030     Xapian::TermIterator t = db->synonym_keys_begin(message);
1031     string reply, prev;
1032     while (t != db->synonym_keys_end(message)) {
1033         const string& term = *t;
1034         size_t reuse = common_prefix_length(prev, term, 255);
1035         reply.append(1, char(reuse));
1036         pack_uint(reply, term.size() - reuse);
1037         reply.append(term, reuse, string::npos);
1038         prev = term;
1039         ++t;
1040     }
1041     send_message(REPLY_SYNONYMKEYLIST, reply);
1042 }
1043
1044 void
1045 RemoteServer::msg_addsynonym(const string& message)
1046 {
1047     if (!wdb)
1048         throw_read_only();
1049     const char* p = message.data();
1050     const char* p_end = p + message.size();
1051     // Get the term
1052     string term;
1053     if (!unpack_string(&p, p_end, term)) {
1054         throw Xapian::NetworkError("Bad MSG_ADDSYNONYM");
1055     }
1056     wdb->add_synonym(term, string(p, p_end - p));
1057     send_message(REPLY_DONE, {});
1058 }
1059
1060 void
1061 RemoteServer::msg_removesynonym(const string& message)
1062 {
1063     if (!wdb)
1064         throw_read_only();
1065     const char* p = message.data();
1066     const char* p_end = p + message.size();
1067     // Get the term
1068     string term;
1069     if (!unpack_string(&p, p_end, term)) {
1070         throw Xapian::NetworkError("Bad MSG_REMOVESYNONYM");
1071     }
1072     wdb->remove_synonym(term, string(p, p_end - p));
1073     send_message(REPLY_DONE, {});
1074 }
1075
1076 void
1077 RemoteServer::msg_clearsynonyms(const string& message)
1078 {
1079     if (!wdb)
1080         throw_read_only();
1081     wdb->clear_synonyms(message);
1082     send_message(REPLY_DONE, {});
1083 }