/* Copyright (c) 2010-2018 Dovecot authors, see the included COPYING file */

/*
   Handshaking:

   Incoming director connections send:

   VERSION
   ME
   <wait for DONE from remote handshake>
   DONE
   <make this connection our "left" connection, potentially disconnecting
   another one>

   Outgoing director connections send:

   VERSION
   ME
   [0..n] DIRECTOR
   HOST-HAND-START
   [0..n] HOST
   HOST-HAND-END
   [0..n] USER
   <possibly other non-handshake commands between USERs>
   DONE
   <wait for DONE from remote>
   <make this connection our "right" connection, potentially disconnecting
   another one>
*/

#include "lib.h"
#include "ioloop.h"
#include "array.h"
#include "net.h"
#include "istream.h"
#include "ostream.h"
#include "str.h"
#include "strescape.h"
#include "time-util.h"
#include "master-service.h"
#include "mail-host.h"
#include "director.h"
#include "director-host.h"
#include "director-request.h"
#include "director-connection.h"

#include <unistd.h>
#include <sys/time.h>
#include <sys/resource.h>

#define MAX_INBUF_SIZE 1024
#define OUTBUF_FLUSH_THRESHOLD (1024*128)
/* Max time to wait for connect() to finish before aborting */
#define DIRECTOR_CONNECTION_CONNECT_TIMEOUT_MSECS (10*1000)
/* Max idling time before "ME" command must have been received,
   or we'll disconnect. */
#define DIRECTOR_CONNECTION_ME_TIMEOUT_MSECS (10*1000)
/* Max time to wait for USERs in handshake to be sent. With a lot of users the
   kernel may quickly eat up everything we send, while the receiver is busy
   parsing the data. */
#define DIRECTOR_CONNECTION_SEND_USERS_TIMEOUT_MSECS (30*1000)
/* Max idling time before "DONE" command must have been received,
   or we'll disconnect. Use a slightly larger value than for _SEND_USERS_ so
   that we'll get a better error if the sender decides to disconnect. */
#define DIRECTOR_CONNECTION_DONE_TIMEOUT_MSECS (40*1000)
/* How long to wait to send PING when connection is idle */
#define DIRECTOR_CONNECTION_PING_INTERVAL_MSECS (15*1000)
/* How long to wait before sending PING while waiting for SYNC reply */
#define DIRECTOR_CONNECTION_PING_SYNC_INTERVAL_MSECS 1000
/* Log a warning if PING reply or PONG response takes longer than this */
#define DIRECTOR_CONNECTION_PINGPONG_WARN_MSECS (5*1000)
/* If outgoing director connection exists for less than this many seconds,
   mark the host as failed so we won't try to reconnect to it immediately */
#define DIRECTOR_SUCCESS_MIN_CONNECT_SECS 40
/* If USER request doesn't have a timestamp, user isn't refreshed if it was
   already refreshed director_user_expire/4 seconds ago. This value is the
   hardcoded maximum for that value. */
#define DIRECTOR_SKIP_RECENT_REFRESH_MAX_SECS 15
#define DIRECTOR_RECONNECT_AFTER_WRONG_CONNECT_MSECS 1000
#define DIRECTOR_WAIT_DISCONNECT_SECS 10
#define DIRECTOR_HANDSHAKE_WARN_SECS 29
#define DIRECTOR_HANDSHAKE_BYTES_LOG_MIN_SECS (60*30)
#define DIRECTOR_MAX_SYNC_SEQ_DUPLICATES 4
/* If we receive SYNCs with a timestamp this many seconds higher than the last
   valid received SYNC timestamp, assume that we lost the director's restart
   notification and reset the last_sync_seq */
#define DIRECTOR_SYNC_STALE_TIMESTAMP_RESET_SECS (60*2)
#define DIRECTOR_MAX_CLOCK_DIFF_WARN_SECS 1
/* How many USER entries to send during handshake before going back to ioloop
   to see if there's other work to be done as well. */
#define DIRECTOR_HANDSHAKE_MAX_USERS_SENT_PER_FLUSH 10000

#define CMD_IS_USER_HANDSHAKE(minor_version, args) \
	((minor_version) < DIRECTOR_VERSION_HANDSHAKE_U_CMD && \
	 str_array_length(args) > 2)

#define DIRECTOR_OPT_CONSISTENT_HASHING "consistent-hashing"

struct director_connection {
	int refcount;
	struct director *dir;
	struct event *event;
	char *name;
	struct timeval created, connected_time, me_received_time;
	struct timeval connected_user_cpu;
	unsigned int minor_version;

	struct timeval last_input, last_output;
	size_t peak_bytes_buffered;

	struct timeval ping_sent_time;
	size_t ping_sent_buffer_size;
	struct timeval ping_sent_user_cpu;
	uoff_t ping_sent_input_offset, ping_sent_output_offset;
	unsigned int last_ping_msecs;

	/* for incoming connections the director host isn't known until
	   ME-line is received */
	struct director_host *host;
	/* this is set only for wrong connections: */
	struct director_host *connect_request_to;

	int fd;
	struct io *io;
	struct istream *input;
	struct ostream *output;
	struct timeout *to_disconnect, *to_ping, *to_pong;

	struct director_user_iter *user_iter;
	unsigned int users_received, handshake_users_received;
	unsigned int handshake_users_sent;

	/* set during command execution */
	const char *cur_cmd, *const *cur_args;

	bool in:1;
	bool connected:1;
	bool version_received:1;
	bool me_received:1;
	bool handshake_received:1;
	bool ignore_host_events:1;
	bool handshake_sending_hosts:1;
	bool ping_waiting:1;
	bool synced:1;
	bool wrong_host:1;
	bool verifying_left:1;
	bool users_unsorted:1;
	bool connected_user_cpu_set:1;
};

static bool director_connection_unref(struct director_connection *conn);
static void director_finish_sending_handshake(struct director_connection *conn);
static void director_connection_disconnected(struct director_connection **conn,
					     const char *reason);
static void director_connection_reconnect(struct director_connection **conn,
					  const char *reason);
static void
director_connection_log_disconnect(struct director_connection *conn, int err,
				   const char *errstr);
static int director_connection_send_done(struct director_connection *conn);

static void
director_connection_set_name(struct director_connection *conn, const char *name)
{
	char *old_name = conn->name;
	conn->name = i_strdup(name);
	i_free(old_name);

	event_set_append_log_prefix(conn->event,
		t_strdup_printf("director(%s): ", conn->name));
}

static void ATTR_FORMAT(2, 3)
director_cmd_error(struct director_connection *conn, const char *fmt, ...)
{
	va_list args;

	va_start(args, fmt);
	e_error(conn->event, "Command %s: %s (input: %s)",
		conn->cur_cmd, t_strdup_vprintf(fmt, args),
		t_strarray_join(conn->cur_args, "\t"));
	va_end(args);

	if (conn->host != NULL)
		conn->host->last_protocol_failure = ioloop_time;
}

static void
director_connection_append_stats(struct director_connection *conn, string_t *str)
{
	struct rusage usage;

	str_printfa(str, "bytes in=%"PRIuUOFF_T", bytes out=%"PRIuUOFF_T,
		    conn->input->v_offset, conn->output->offset);
	str_printfa(str, ", %u+%u USERs received",
		    conn->handshake_users_received, conn->users_received);
	if (conn->handshake_users_sent > 0) {
		str_printfa(str, ", %u USERs sent in handshake",
			    conn->handshake_users_sent);
	}
	if (conn->last_input.tv_sec > 0) {
		int input_msecs = timeval_diff_msecs(&ioloop_timeval,
						     &conn->last_input);
		str_printfa(str, ", last input %u.%03u s ago",
			    input_msecs/1000, input_msecs%1000);
	}
	if (conn->last_output.tv_sec > 0) {
		int output_msecs = timeval_diff_msecs(&ioloop_timeval,
						      &conn->last_output);
		str_printfa(str, ", last output %u.%03u s ago",
			    output_msecs/1000, output_msecs%1000);
	}
	if (conn->connected) {
		int connected_msecs = timeval_diff_msecs(&ioloop_timeval,
							 &conn->connected_time);
		str_printfa(str, ", connected %u.%03u s ago",
			    connected_msecs/1000, connected_msecs%1000);
	}
	if (o_stream_get_buffer_used_size(conn->output) > 0) {
		str_printfa(str, ", %zu bytes in output buffer",
			    o_stream_get_buffer_used_size(conn->output));
	}
	str_printfa(str, ", %zu peak output buffer size",
		    conn->peak_bytes_buffered);
	if (conn->connected_user_cpu_set &&
	    getrusage(RUSAGE_SELF, &usage) == 0) {
		/* this isn't measuring the CPU usage used by the connection
		   itself, but it can still be a useful measurement */
		int diff = timeval_diff_msecs(&usage.ru_utime,
					      &conn->connected_user_cpu);
		str_printfa(str, ", %d.%03d CPU secs since connected",
			    diff / 1000, diff % 1000);
	}
}

static void
director_connection_init_timeout(struct director_connection *conn)
{
	struct timeval start_time;
	string_t *reason = t_str_new(128);

	if (!conn->connected) {
		start_time = conn->created;
		str_append(reason, "Connect timed out");
	} else if (!conn->me_received) {
		start_time = conn->connected_time;
		str_append(reason, "Handshaking ME timed out");
	} else if (!conn->in) {
		start_time = conn->me_received_time;
		str_append(reason, "Sending handshake timed out");
	} else {
		start_time = conn->me_received_time;
		str_append(reason, "Handshaking DONE timed out");
	}
	int msecs = timeval_diff_msecs(&ioloop_timeval, &start_time);
	str_printfa(reason, " (%u.%03u secs, ", msecs/1000, msecs%1000);
	director_connection_append_stats(conn, reason);
	str_append_c(reason, ')');

	e_error(conn->event, "%s", str_c(reason));
	director_connection_disconnected(&conn, "Handshake timeout");
}

static void
director_connection_set_ping_timeout(struct director_connection *conn)
{
	unsigned int msecs;

	msecs = conn->synced || !conn->handshake_received ?
		DIRECTOR_CONNECTION_PING_INTERVAL_MSECS :
		DIRECTOR_CONNECTION_PING_SYNC_INTERVAL_MSECS;

	timeout_remove(&conn->to_ping);
	conn->to_ping = timeout_add(msecs, director_connection_ping, conn);
}

static void director_connection_wait_timeout(struct director_connection *conn)
{
	director_connection_log_disconnect(conn, ETIMEDOUT, "");
	director_connection_deinit(&conn,
		"Timeout waiting for disconnect after CONNECT");
}

static void director_connection_send_connect(struct director_connection *conn,
					     struct director_host *host)
{
	const char *connect_str;

	if (conn->to_disconnect != NULL)
		return;

	connect_str = t_strdup_printf("CONNECT\t%s\t%u\n",
				      host->ip_str, host->port);
	director_connection_send(conn, connect_str);
	o_stream_uncork(conn->output);

	/* wait for a while for the remote to disconnect, so it will hopefully
	   see our CONNECT command. we'll also log the warning later to avoid
	   multiple log lines about it. */
	conn->connect_request_to = host;
	director_host_ref(conn->connect_request_to);

	conn->to_disconnect =
		timeout_add(DIRECTOR_WAIT_DISCONNECT_SECS*1000,
			    director_connection_wait_timeout, conn);
}

static void director_connection_assigned(struct director_connection *conn)
{
	struct director *dir = conn->dir;

	if (dir->left != NULL && dir->right != NULL) {
		/* we're connected to both directors. see if the ring is
		   finished by sending a SYNC. if we get it back, it's done. */
		dir->sync_seq++;
		director_set_ring_unsynced(dir);
		director_sync_send(dir, dir->self_host, dir->sync_seq,
				   DIRECTOR_VERSION_MINOR, ioloop_time,
				   mail_hosts_hash(dir->mail_hosts));
	}
	director_connection_set_ping_timeout(conn);
}

static bool director_connection_assign_left(struct director_connection *conn)
{
	struct director *dir = conn->dir;

	i_assert(conn->in);
	i_assert(dir->left != conn);

	/* make sure this is the correct incoming connection */
	if (conn->host->self) {
		e_error(conn->event, "Connection from self, dropping");
		return FALSE;
	} else if (dir->left == NULL) {
		/* no conflicts yet */
	} else if (dir->left->host == conn->host) {
		e_warning(conn->event,
			  "Replacing left director connection %s with %s",
			  dir->left->host->name, conn->host->name);
		director_connection_deinit(&dir->left, t_strdup_printf(
			"Replacing with %s", conn->host->name));
	} else if (dir->left->verifying_left) {
		/* we're waiting to verify if our current left is still
		   working. if we don't receive a PONG, the current left
		   gets disconnected and a new left gets assigned. if we do
		   receive a PONG, we'll wait until the current left
		   disconnects us and then reassign the new left. */
		return TRUE;
	} else if (director_host_cmp_to_self(dir->left->host, conn->host,
					     dir->self_host) < 0) {
		/* the old connection is the correct one.
		   refer the client there (FIXME: do we ever get here?) */
		director_connection_send_connect(conn, dir->left->host);
		return TRUE;
	} else {
		/* this new connection is the correct one, but wait until the
		   old connection gets disconnected before using this one.
		   that guarantees that the director inserting itself into
		   the ring has finished handshaking its left side, so the
		   switch will be fast. */
		return TRUE;
	}
	dir->left = conn;
	director_connection_set_name(conn,
		t_strdup_printf("%s/left", conn->host->name));
	director_connection_assigned(conn);
	return TRUE;
}

static void director_assign_left(struct director *dir)
{
	struct director_connection *conn;

	array_foreach_elem(&dir->connections, conn) {
		if (conn->in && conn->handshake_received &&
		    conn->to_disconnect == NULL && conn != dir->left) {
			/* either use this or disconnect it */
			if (!director_connection_assign_left(conn)) {
				/* we don't want this */
				director_connection_deinit(&conn,
					"Unwanted incoming connection");
				director_assign_left(dir);
				break;
			}
		}
	}
}

static bool director_has_outgoing_connections(struct director *dir)
{
	struct director_connection *conn;

	array_foreach_elem(&dir->connections, conn) {
		if (!conn->in && conn->to_disconnect == NULL)
			return TRUE;
	}
	return FALSE;
}

static void director_send_delayed_syncs(struct director *dir)
{
	struct director_host *host;

	i_assert(dir->right != NULL);

	e_debug(dir->right->event, "Sending delayed SYNCs");
	array_foreach_elem(&dir->dir_hosts, host) {
		if (host->delayed_sync_seq == 0)
			continue;

		director_sync_send(dir, host, host->delayed_sync_seq,
				   host->delayed_sync_minor_version,
				   host->delayed_sync_timestamp,
				   host->delayed_sync_hosts_hash);
		host->delayed_sync_seq = 0;
	}
}

static bool director_connection_assign_right(struct director_connection *conn)
{
	struct director *dir = conn->dir;

	i_assert(!conn->in);

	if (dir->right != NULL) {
		/* see if we should disconnect or keep the existing
		   connection. */
		if (director_host_cmp_to_self(conn->host, dir->right->host,
					      dir->self_host) <= 0) {
			/* the old connection is the correct one */
			e_warning(conn->event,
				  "Aborting incorrect outgoing connection to %s "
				  "(already connected to correct one: %s)",
				  conn->host->name, dir->right->host->name);
			conn->wrong_host = TRUE;
			return FALSE;
		}
		e_warning(conn->event,
			  "Replacing right director connection %s with %s",
			  dir->right->host->name, conn->host->name);
		director_connection_deinit(&dir->right, t_strdup_printf(
			"Replacing with %s", conn->host->name));
	}
	dir->right = conn;
	director_connection_set_name(conn,
		t_strdup_printf("%s/right", conn->host->name));
	director_connection_assigned(conn);
	director_send_delayed_syncs(dir);
	return TRUE;
}

static bool
director_args_parse_ip_port(struct director_connection *conn,
			    const char *const *args,
			    struct ip_addr *ip_r, in_port_t *port_r)
{
	if (args[0] == NULL || args[1] == NULL) {
		director_cmd_error(conn, "Missing IP+port parameters");
		return FALSE;
	}
	if (net_addr2ip(args[0], ip_r) < 0) {
		director_cmd_error(conn, "Invalid IP address: %s", args[0]);
		return FALSE;
	}
	if (net_str2port(args[1], port_r) < 0) {
		director_cmd_error(conn, "Invalid port: %s", args[1]);
		return FALSE;
	}
	return TRUE;
}

static bool director_cmd_me(struct director_connection *conn,
			    const char *const *args)
{
	struct director *dir = conn->dir;
	const char *connect_str;
	struct ip_addr ip;
	in_port_t port;
	time_t next_comm_attempt;

	if (!director_args_parse_ip_port(conn, args, &ip, &port))
		return FALSE;
	if (conn->me_received) {
		director_cmd_error(conn, "Duplicate ME");
		return FALSE;
	}

	if (!conn->in && (!net_ip_compare(&conn->host->ip, &ip) ||
			  conn->host->port != port)) {
		e_error(conn->event,
			"Remote director thinks it's someone else "
			"(connected to %s:%u, remote says it's %s:%u)",
			conn->host->ip_str, conn->host->port,
			net_ip2addr(&ip), port);
		return FALSE;
	}
	conn->me_received = TRUE;
	conn->me_received_time = ioloop_timeval;

	if (args[2] != NULL) {
		time_t remote_time;
		int diff;

		if (str_to_time(args[2], &remote_time) < 0) {
			director_cmd_error(conn, "Invalid ME timestamp");
			return FALSE;
		}
		diff = ioloop_time - remote_time;
		if (diff > DIRECTOR_MAX_CLOCK_DIFF_WARN_SECS ||
		    (diff < 0 && -diff > DIRECTOR_MAX_CLOCK_DIFF_WARN_SECS)) {
			e_warning(conn->event,
				  "Director %s clock differs from ours by %d secs",
				  conn->name, diff);
		}
	}

	timeout_remove(&conn->to_ping);
	if (conn->in) {
		conn->to_ping = timeout_add(DIRECTOR_CONNECTION_DONE_TIMEOUT_MSECS,
					    director_connection_init_timeout, conn);
	} else {
		conn->to_ping = timeout_add(DIRECTOR_CONNECTION_SEND_USERS_TIMEOUT_MSECS,
					    director_connection_init_timeout, conn);
	}

	if (!conn->in)
		return TRUE;

	/* Incoming connection:

	   a) we don't have an established ring yet. make sure we're connecting
	   to our right side (which might become our left side).

	   b) it's our current "left" connection. the previous connection
	   is most likely dead.

	   c) we have an existing ring. tell our current "left" to connect to
	   it with CONNECT command.

	   d) the incoming connection doesn't belong to us at all, refer it
	   elsewhere with CONNECT. however, before disconnecting it verify
	   first that our left side is actually still functional.
	*/
	i_assert(conn->host == NULL);
	conn->host = director_host_get(dir, &ip, port);
	/* the host shouldn't be removed at this point, but if for some
	   reason it is we don't want to crash */
	conn->host->removed = FALSE;
	director_host_ref(conn->host);
	/* make sure we don't keep old sequence values across restarts */
	director_host_restarted(conn->host);

	next_comm_attempt = conn->host->last_protocol_failure +
		DIRECTOR_PROTOCOL_FAILURE_RETRY_SECS;
	if (next_comm_attempt > ioloop_time) {
		/* the director recently sent invalid protocol data,
		   don't try retrying yet */
		e_error(conn->event,
			"Remote sent invalid protocol data recently, "
			"waiting %u secs before allowing further communication",
			(unsigned int)(next_comm_attempt-ioloop_time));
		return FALSE;
	} else if (dir->left == NULL) {
		/* a) - just in case the left is also our right side reset
		   its failed state, so we can connect to it */
		conn->host->last_network_failure = 0;
		if (!director_has_outgoing_connections(dir))
			director_connect(dir, "Connecting to left");
	} else if (dir->left->host == conn->host) {
		/* b) */
		i_assert(dir->left != conn);
		director_connection_deinit(&dir->left,
			"Replacing with new incoming connection");
	} else if (director_host_cmp_to_self(conn->host, dir->left->host,
					     dir->self_host) < 0) {
		/* c) */
		connect_str = t_strdup_printf("CONNECT\t%s\t%u\n",
					      conn->host->ip_str,
					      conn->host->port);
		director_connection_send(dir->left, connect_str);
	} else {
		/* d) */
		dir->left->verifying_left = TRUE;
		director_connection_ping(dir->left);
	}
	return TRUE;
}

static inline bool
user_need_refresh(struct director *dir, struct user *user,
		  time_t timestamp, bool unknown_timestamp)
{
	if (timestamp <= (time_t)user->timestamp) {
		/* we already have this timestamp */
		return FALSE;
	}
	if (unknown_timestamp) {
		/* Old director sent USER command without timestamp. We don't
		   know what it is exactly, but we can assume that it's very
		   close to the current time (which timestamp parameter is
		   already set to). However, try to break USER loops here when
		   director ring latency is >1sec, but below skip_recent_secs
		   by just not refreshing the user. */
		time_t skip_recent_secs =
			I_MIN(dir->set->director_user_expire/4,
			      DIRECTOR_SKIP_RECENT_REFRESH_MAX_SECS);
		if ((time_t)user->timestamp + skip_recent_secs >= timestamp)
			return FALSE;
	}
	return TRUE;
}

static int
director_user_refresh(struct director_connection *conn,
		      unsigned int username_hash, struct mail_host *host,
		      time_t timestamp, bool weak, bool *forced_r,
		      struct user **user_r)
{
	struct director *dir = conn->dir;
	struct user *user;
	bool ret = FALSE, unset_weak_user = FALSE;
	struct user_directory *users = host->tag->users;
	bool unknown_timestamp = (timestamp == (time_t)-1);

	*forced_r = FALSE;

	if (unknown_timestamp) {
		/* Old director version sent USER without timestamp. */
		timestamp = ioloop_time;
	}

	if (timestamp + (time_t)dir->set->director_user_expire <= ioloop_time) {
		/* Ignore this refresh entirely, regardless of whether the
		   user already exists or not. */
		e_debug(conn->event,
			"user refresh: %u has expired timestamp %"PRIdTIME_T,
			  username_hash, timestamp);
		return -1;
	}

	user = user_directory_lookup(users, username_hash);
	if (user == NULL) {
		*user_r = user_directory_add(users, username_hash,
					     host, timestamp);
		(*user_r)->weak = weak;
		e_debug(conn->event, "user refresh: %u added", username_hash);
		return 1;
	}

	if (user->weak) {
		if (!weak) {
			/* removing user's weakness */
			e_debug(conn->event, "user refresh: %u weakness removed",
				username_hash);
			unset_weak_user = TRUE;
			user->weak = FALSE;
			ret = TRUE;
		} else {
			/* weak user marked again as weak */
		}
	} else if (weak &&
		   !user_directory_user_is_recently_updated(users, user)) {
		/* mark the user as weak */
		e_debug(conn->event, "user refresh: %u set weak", username_hash);
		user->weak = TRUE;
		ret = TRUE;
	} else if (weak) {
		e_debug(conn->event,
			"user refresh: %u weak update to %s ignored, "
			"we recently changed it to %s",
			username_hash, host->ip_str,
			user->host->ip_str);
		host = user->host;
		ret = TRUE;
	} else if (user->host == host) {
		/* update to the same host */
	} else if (user_directory_user_is_near_expiring(users, user)) {
		/* host conflict for a user that is already near expiring. we can
		   assume that the other director had already dropped this user
		   and we should have as well. use the new host. */
		e_debug(conn->event, "user refresh: %u is nearly expired, "
			"replacing host %s with %s", username_hash,
			user->host->ip_str, host->ip_str);
		ret = TRUE;
	} else if (USER_IS_BEING_KILLED(user)) {
		/* user is still being moved - ignore conflicting host updates
		   from other directors who don't yet know about the move. */
		e_debug(conn->event, "user refresh: %u is being moved, "
			"preserve its host %s instead of replacing with %s",
			username_hash, user->host->ip_str, host->ip_str);
		host = user->host;
	} else {
		/* non-weak user received a non-weak update with
		   conflicting host. this shouldn't happen. */
		string_t *str = t_str_new(128);

		str_printfa(str, "User hash %u "
			    "is being redirected to two hosts: %s and %s",
			    username_hash, user->host->ip_str, host->ip_str);
		str_printfa(str, " (old_ts=%ld", (long)user->timestamp);

		if (!conn->handshake_received) {
			str_printfa(str, ",handshaking,recv_ts=%ld",
				    (long)timestamp);
		}
		if (USER_IS_BEING_KILLED(user)) {
			if (user->kill_ctx->to_move != NULL)
				str_append(str, ",moving");
			str_printfa(str, ",kill_state=%s",
				    user_kill_state_names[user->kill_ctx->kill_state]);
		}
		str_append_c(str, ')');
		e_error(conn->event, "%s", str_c(str));

		/* we want all the directors to redirect the user to same
		   server, but we don't want two directors fighting over which
		   server it belongs to, so always use the lower IP address */
		if (net_ip_cmp(&user->host->ip, &host->ip) > 0) {
			/* change the host. we'll also need to remove the user
			   from the old host's user_count, because we can't
			   keep track of the user for more than one host.

			   send the updated USER back to the sender as well. */
			*forced_r = TRUE;
		} else {
			/* keep the host */
			host = user->host;
		}
		/* especially IMAP connections can take a long time to die.
		   make sure we kill off the connections in the wrong
		   backends. */
		director_kick_user_hash(dir, dir->self_host, NULL,
					username_hash, &host->ip);
		ret = TRUE;
	}
	if (user->host != host) {
		user->host->user_count--;
		user->host = host;
		user->host->user_count++;
		ret = TRUE;
	}
	/* Update user's timestamp if it's higher than the current one. Note
	   that we'll preserve the original timestamp. This is important when
	   the director ring is slow and a single USER can traverse through
	   the ring more than a second. We don't want to get into a loop where
	   the same USER goes through the ring forever. */
	if (user_need_refresh(dir, user, timestamp, unknown_timestamp)) {
		/* NOTE: This makes the users list somewhat out-of-order.
		   It's not a big problem - most likely it's only a few seconds
		   difference. The worst that can happen is that some users
		   take up memory that should have been freed already. */
		e_debug(conn->event, "user refresh: %u refreshed timestamp from %u to %"PRIdTIME_T,
			username_hash, user->timestamp, timestamp);
		user_directory_refresh(users, user);
		user->timestamp = timestamp;
		ret = TRUE;
	} else {
		e_debug(conn->event, "user refresh: %u ignored timestamp %"PRIdTIME_T" (we have %u)",
			username_hash, timestamp, user->timestamp);
	}

	if (unset_weak_user) {
		/* user is no longer weak. handle pending requests for
		   this user if there are any */
		director_set_state_changed(conn->dir);
	}

	*user_r = user;
	return ret ? 1 : 0;
}

static bool
director_handshake_cmd_user(struct director_connection *conn,
			    const char *const *args)
{
	unsigned int username_hash, timestamp;
	struct ip_addr ip;
	struct mail_host *host;
	struct user *user;
	bool weak, forced;

	if (str_array_length(args) < 3 ||
	    str_to_uint(args[0], &username_hash) < 0 ||
	    net_addr2ip(args[1], &ip) < 0 ||
	    str_to_uint(args[2], &timestamp) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}
	weak = args[3] != NULL && args[3][0] == 'w';
	conn->handshake_users_received++;

	host = mail_host_lookup(conn->dir->mail_hosts, &ip);
	if (host == NULL) {
		e_error(conn->event, "USER used unknown host %s in handshake",
			args[1]);
		return FALSE;
	}

	if ((time_t)timestamp > ioloop_time) {
		/* The other director's clock seems to be into the future
		   compared to us. Don't set any of our users' timestamps into
		   future though. It's most likely only 1 second difference. */
		timestamp = ioloop_time;
	}
	conn->dir->num_incoming_requests++;
	if (director_user_refresh(conn, username_hash, host,
				  timestamp, weak, &forced, &user) < 0) {
		/* user expired - ignore */
		return TRUE;
	}
	/* Possibilities:

	   a) The user didn't exist yet, and it was added with the given
	   timestamp.

	   b) The user existed, but with an older timestamp. The timestamp
	   wasn't yet updated, so do it here below.

	   c) The user existed with a newer timestamp. This is either because
	   we already received a non-handshake USER update for this user, or
	   our director saw a login for this user. Ignore this update.

	   (We never want to change the user's timestamp to be older, because
	   that could result in directors going to a loop fighting each others
	   over a flipping timestamp.) */
	if (user->timestamp < timestamp)
		user->timestamp = timestamp;
	/* always sort users after handshaking to make sure the order
	   is correct */
	conn->users_unsorted = TRUE;
	return TRUE;
}

static bool
director_cmd_user(struct director_connection *conn,
		  const char *const *args)
{
	unsigned int username_hash;
	struct ip_addr ip;
	struct mail_host *host;
	struct user *user;
	bool forced;
	time_t timestamp = (time_t)-1;

	if (str_array_length(args) < 2 ||
	    str_to_uint(args[0], &username_hash) < 0 ||
	    net_addr2ip(args[1], &ip) < 0 ||
	    (args[2] != NULL && str_to_time(args[2], &timestamp) < 0)) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}

	/* could this before it's potentially ignored */
	conn->dir->num_incoming_requests++;

	conn->users_received++;
	host = mail_host_lookup(conn->dir->mail_hosts, &ip);
	if (host == NULL) {
		/* we probably just removed this host. */
		return TRUE;
	}

	if (director_user_refresh(conn, username_hash,
				  host, timestamp, FALSE, &forced, &user) > 0) {
		/* user changed - forward the USER in the ring */
		struct director_host *src_host =
			forced ? conn->dir->self_host : conn->host;
		i_assert(!user->weak);
		director_update_user(conn->dir, src_host, user);
	}
	return TRUE;
}

static bool director_cmd_director(struct director_connection *conn,
				  const char *const *args)
{
	struct director_host *host;
	struct ip_addr ip;
	in_port_t port;
	bool log_add = FALSE;

	if (!director_args_parse_ip_port(conn, args, &ip, &port))
		return FALSE;

	host = director_host_lookup(conn->dir, &ip, port);
	if (host != NULL) {
		if (host == conn->dir->self_host) {
			/* ignore updates to ourself */
			return TRUE;
		}
		if (host->removed) {
			/* ignore re-adds of removed directors */
			return TRUE;
		}

		/* already have this. just reset its last_network_failure
		   timestamp, since it might be up now, but only if this
		   isn't part of the handshake. (if it was, reseting the
		   timestamp could cause us to rapidly keep trying to connect
		   to it) */
		if (conn->handshake_received)
			host->last_network_failure = 0;
		/* it also may have been restarted, reset its state */
		director_host_restarted(host);
	} else {
		/* save the director and forward it */
		host = director_host_add(conn->dir, &ip, port);
		log_add = TRUE;
	}
	/* just forward this to the entire ring until it reaches back to
	   itself. some hosts may see this twice, but that's the only way to
	   guarantee that it gets seen by everyone. resetting the host multiple
	   times may cause us to handle its commands multiple times, but the
	   commands can handle that. however, we need to also handle a
	   situation where the added director never comes back - we don't want
	   to send the director information in a loop forever. */
	if (conn->dir->right != NULL &&
	    director_host_cmp_to_self(host, conn->dir->right->host,
				      conn->dir->self_host) > 0) {
		e_debug(conn->event,
			"Received DIRECTOR update for a host where we should be connected to. "
			"Not forwarding it since it's probably crashed.");
	} else {
		director_notify_ring_added(host,
			director_connection_get_host(conn), log_add);
	}
	return TRUE;
}

static bool director_cmd_director_remove(struct director_connection *conn,
					 const char *const *args)
{
	struct director_host *host;
	struct ip_addr ip;
	in_port_t port;

	if (!director_args_parse_ip_port(conn, args, &ip, &port))
		return FALSE;

	host = director_host_lookup(conn->dir, &ip, port);
	if (host != NULL && !host->removed)
		director_ring_remove(host, director_connection_get_host(conn));
	return TRUE;
}

static bool
director_cmd_host_hand_start(struct director_connection *conn,
			     const char *const *args)
{
	const ARRAY_TYPE(mail_host) *hosts;
	struct mail_host *const *hostp;
	unsigned int remote_ring_completed;

	if (args[0] == NULL ||
	    str_to_uint(args[0], &remote_ring_completed) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}

	if (remote_ring_completed != 0 && !conn->dir->ring_handshaked) {
		/* clear everything we have and use only what remote sends us */
		e_debug(conn->event, "We're joining a ring - replace all hosts");
		hosts = mail_hosts_get(conn->dir->mail_hosts);
		while (array_count(hosts) > 0) {
			hostp = array_front(hosts);
			director_remove_host(conn->dir, NULL, NULL, *hostp);
		}
	} else if (remote_ring_completed == 0 && conn->dir->ring_handshaked) {
		/* ignore whatever remote sends */
		e_debug(conn->event, "Remote is joining our ring - "
			"ignore all remote HOSTs");
		conn->ignore_host_events = TRUE;
	} else {
		e_debug(conn->event, "Merge rings' hosts");
	}
	conn->handshake_sending_hosts = TRUE;
	return TRUE;
}

static int
director_cmd_is_seen_full(struct director_connection *conn,
			  const char *const **_args, unsigned int *seq_r,
			  struct director_host **host_r)
{
	const char *const *args = *_args;
	struct ip_addr ip;
	in_port_t port;
	unsigned int seq;
	struct director_host *host;

	if (str_array_length(args) < 3 ||
	    net_addr2ip(args[0], &ip) < 0 ||
	    net_str2port(args[1], &port) < 0 ||
	    str_to_uint(args[2], &seq) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return -1;
	}
	*_args = args + 3;
	*seq_r = seq;

	host = director_host_lookup(conn->dir, &ip, port);
	if (host == NULL || host->removed) {
		/* director is already gone, but we can't be sure if this
		   command was sent everywhere. re-send it as if it was from
		   ourself. */
		*host_r = NULL;
	} else {
		*host_r = host;
		if (seq <= host->last_seq) {
			/* already seen this */
			return 1;
		}
		host->last_seq = seq;
	}
	return 0;
}

static int
director_cmd_is_seen(struct director_connection *conn,
		     const char *const **_args,
		     struct director_host **host_r)
{
	unsigned int seq;

	return director_cmd_is_seen_full(conn, _args, &seq, host_r);
}

static bool
director_cmd_user_weak(struct director_connection *conn,
		       const char *const *args)
{
	struct director_host *dir_host;
	struct ip_addr ip;
	unsigned int username_hash;
	struct mail_host *host;
	struct user *user;
	struct director_host *src_host = conn->host;
	bool weak = TRUE, weak_forward = FALSE, forced;
	int ret;

	/* note that unlike other commands we don't want to just ignore
	   duplicate commands */
	if ((ret = director_cmd_is_seen(conn, &args, &dir_host)) < 0)
		return FALSE;

	/* could this before it's potentially ignored */
	conn->dir->num_incoming_requests++;

	if (str_array_length(args) != 2 ||
	    str_to_uint(args[0], &username_hash) < 0 ||
	    net_addr2ip(args[1], &ip) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}

	host = mail_host_lookup(conn->dir->mail_hosts, &ip);
	if (host == NULL) {
		/* we probably just removed this host. */
		return TRUE;
	}

	if (ret == 0) {
		/* First time we're seeing this - forward it to others also.
		   We'll want to do it even if the user was already marked as
		   weak, because otherwise if two directors mark the user weak
		   at the same time both the USER-WEAK notifications reach
		   only half the directors until they collide and neither one
		   finishes going through the whole ring marking the user
		   non-weak. */
		weak_forward = TRUE;
	} else if (dir_host == conn->dir->self_host) {
		/* We originated this USER-WEAK request. The entire ring has seen
		   it and there weren't any conflicts. Make the user non-weak. */
		e_debug(conn->event,
			"user refresh: %u Our USER-WEAK seen by the entire ring",
			username_hash);
		src_host = conn->dir->self_host;
		weak = FALSE;
	} else {
		/* The original USER-WEAK sender will send a new non-weak USER
		   update saying what really happened. We'll still need to forward
		   this around the ring to the origin so it also knows it has
		   travelled through the ring. */
		e_debug(conn->event,
			"user refresh: %u Remote USER-WEAK from %s seen by the entire ring, ignoring",
			username_hash, dir_host->ip_str);
		weak_forward = TRUE;
	}

	ret = director_user_refresh(conn, username_hash,
				    host, ioloop_time, weak, &forced, &user);
	/* user is refreshed with ioloop_time, it can't be expired already */
	i_assert(ret >= 0);
	if (ret > 0 || weak_forward) {
		/* user changed, or we've decided that we need to forward
		   the weakness notification to the rest of the ring even
		   though we already knew it. */
		if (forced)
			src_host = conn->dir->self_host;
		if (!user->weak)
			director_update_user(conn->dir, src_host, user);
		else {
			director_update_user_weak(conn->dir, src_host, conn,
						  dir_host, user);
		}
	}
	return TRUE;
}

static bool ATTR_NULL(3)
director_cmd_host_int(struct director_connection *conn, const char *const *args,
		      struct director_host *dir_host)
{
	struct director_host *src_host = conn->host;
	struct mail_host *host;
	struct ip_addr ip;
	const char *tag = "", *host_tag, *hostname = NULL;
	unsigned int arg_count, vhost_count;
	bool update, down = FALSE, tag_changed = FALSE;
	time_t last_updown_change = 0;

	arg_count = str_array_length(args);
	if (arg_count < 2 ||
	    net_addr2ip(args[0], &ip) < 0 ||
	    str_to_uint(args[1], &vhost_count) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}
	if (arg_count >= 3)
		tag = args[2];
	if (arg_count >= 4) {
		if ((args[3][0] != 'D' && args[3][0] != 'U') ||
		    str_to_time(args[3]+1, &last_updown_change) < 0) {
			director_cmd_error(conn, "Invalid updown parameters");
			return FALSE;
		}
		down = args[3][0] == 'D';
	}
	if (arg_count >= 5)
		hostname = args[4];
	if (conn->ignore_host_events) {
		/* remote is sending hosts in a handshake, but it doesn't have
		   a completed ring and we do. */
		i_assert(conn->handshake_sending_hosts);
		return TRUE;
	}
	if (tag[0] != '\0' && conn->minor_version < DIRECTOR_VERSION_TAGS_V2) {
		director_cmd_error(conn, "Received a host tag from older director version with incompatible tagging support");
		return FALSE;
	}

	host = mail_host_lookup(conn->dir->mail_hosts, &ip);
	if (host == NULL) {
		host = mail_host_add_hostname(conn->dir->mail_hosts,
					      hostname, &ip, tag);
		update = TRUE;
	} else {
		host_tag = mail_host_get_tag(host);
		tag_changed = strcmp(tag, host_tag) != 0;
		update = host->vhost_count != vhost_count ||
			host->down != down || tag_changed;

		if (update && host->desynced) {
			string_t *str = t_str_new(128);

			str_printfa(str, "Host %s is being updated before previous update had finished (",
				  host->ip_str);
			if (host->down != down &&
			    host->last_updown_change > last_updown_change) {
				/* our host has a newer change. preserve it. */
				down = host->down;
			}
			if (host->down != down) {
				if (host->down)
					str_append(str, "down -> up");
				else
					str_append(str, "up -> down");
			}
			if (host->vhost_count != vhost_count) {
				if (host->down != down)
					str_append(str, ", ");
				str_printfa(str, "vhosts %u -> %u",
					    host->vhost_count, vhost_count);
			}
			str_append(str, ") - ");

			vhost_count = I_MIN(vhost_count, host->vhost_count);
			str_printfa(str, "setting to state=%s vhosts=%u",
				    down ? "down" : "up", vhost_count);
			e_warning(conn->event, "%s", str_c(str));
			/* make the change appear to come from us, so it
			   reaches the full ring */
			dir_host = NULL;
			src_host = conn->dir->self_host;
		}
		if (update) {
			/* Make sure the host's timestamp never shrinks.
			   Otherwise we might get into a loop where the up/down
			   state keeps switching. */
			last_updown_change = I_MAX(last_updown_change,
						   host->last_updown_change);
		}
	}

	if (update) {
		const char *log_prefix = t_strdup_printf("director(%s): ",
							 conn->name);
		if (tag_changed) {
			e_error(conn->event,
				"Host %s changed tag from '%s' to '%s'",
				host->ip_str,
				mail_host_get_tag(host), tag);
			mail_host_set_tag(host, tag);
		}
		mail_host_set_down(host, down, last_updown_change, log_prefix);
		mail_host_set_vhost_count(host, vhost_count, log_prefix);
		director_update_host(conn->dir, src_host, dir_host, host);
	} else {
		e_debug(conn->event,
			"Ignoring host %s update vhost_count=%u "
			"down=%d last_updown_change=%ld (hosts_hash=%u)",
			net_ip2addr(&ip), vhost_count, down ? 1 : 0,
			(long)last_updown_change,
			mail_hosts_hash(conn->dir->mail_hosts));
	}
	return TRUE;
}

static bool
director_cmd_host_handshake(struct director_connection *conn,
			    const char *const *args)
{
	return director_cmd_host_int(conn, args, NULL);
}

static bool
director_cmd_host(struct director_connection *conn, const char *const *args)
{
	struct director_host *dir_host;
	int ret;

	if ((ret = director_cmd_is_seen(conn, &args, &dir_host)) != 0)
		return ret > 0;
	return director_cmd_host_int(conn, args, dir_host);
}

static bool
director_cmd_host_remove(struct director_connection *conn,
			 const char *const *args)
{
	struct director_host *dir_host;
	struct mail_host *host;
	struct ip_addr ip;
	int ret;

	if ((ret = director_cmd_is_seen(conn, &args, &dir_host)) != 0)
		return ret > 0;

	if (str_array_length(args) != 1 ||
	    net_addr2ip(args[0], &ip) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}

	host = mail_host_lookup(conn->dir->mail_hosts, &ip);
	if (host != NULL)
		director_remove_host(conn->dir, conn->host, dir_host, host);
	return TRUE;
}

static bool
director_cmd_host_flush(struct director_connection *conn,
			 const char *const *args)
{
	struct director_host *dir_host;
	struct mail_host *host;
	struct ip_addr ip;
	int ret;

	if ((ret = director_cmd_is_seen(conn, &args, &dir_host)) != 0)
		return ret > 0;

	if (str_array_length(args) != 1 ||
	    net_addr2ip(args[0], &ip) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}

	host = mail_host_lookup(conn->dir->mail_hosts, &ip);
	if (host != NULL)
		director_flush_host(conn->dir, conn->host, dir_host, host);
	return TRUE;
}

static bool
director_cmd_user_move(struct director_connection *conn,
		       const char *const *args)
{
	struct director_host *dir_host;
	struct mail_host *host;
	struct ip_addr ip;
	unsigned int username_hash;
	int ret;

	if ((ret = director_cmd_is_seen(conn, &args, &dir_host)) != 0)
		return ret > 0;

	if (str_array_length(args) != 2 ||
	    str_to_uint(args[0], &username_hash) < 0 ||
	    net_addr2ip(args[1], &ip) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}

	host = mail_host_lookup(conn->dir->mail_hosts, &ip);
	if (host != NULL) {
		director_move_user(conn->dir, conn->host, dir_host,
				   username_hash, host);
	}
	return TRUE;
}

static bool
director_cmd_user_kick(struct director_connection *conn,
		       const char *const *args)
{
	struct director_host *dir_host;
	int ret;

	if ((ret = director_cmd_is_seen(conn, &args, &dir_host)) != 0)
		return ret > 0;

	if (str_array_length(args) != 1) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}

	director_kick_user(conn->dir, conn->host, dir_host, args[0]);
	return TRUE;
}

static bool
director_cmd_user_kick_alt(struct director_connection *conn,
			   const char *const *args)
{
	struct director_host *dir_host;
	int ret;

	if ((ret = director_cmd_is_seen(conn, &args, &dir_host)) != 0)
		return ret > 0;

	if (str_array_length(args) != 2) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}

	director_kick_user_alt(conn->dir, conn->host, dir_host, args[0], args[1]);
	return TRUE;
}

static bool
director_cmd_user_kick_hash(struct director_connection *conn,
			    const char *const *args)
{
	struct director_host *dir_host;
	unsigned int username_hash;
	struct ip_addr except_ip;
	int ret;

	if ((ret = director_cmd_is_seen(conn, &args, &dir_host)) != 0)
		return ret > 0;

	if (str_array_length(args) != 2 ||
	    str_to_uint(args[0], &username_hash) < 0 ||
	    net_addr2ip(args[1], &except_ip) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}

	director_kick_user_hash(conn->dir, conn->host, dir_host,
				username_hash, &except_ip);
	return TRUE;
}

static bool
director_cmd_user_killed(struct director_connection *conn,
			 const char *const *args)
{
	unsigned int username_hash;

	if (str_array_length(args) != 1 ||
	    str_to_uint(args[0], &username_hash) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}

	director_user_killed(conn->dir, username_hash);
	return TRUE;
}

static bool
director_cmd_user_killed_everywhere(struct director_connection *conn,
				    const char *const *args)
{
	struct director_host *dir_host;
	unsigned int seq, username_hash;
	int ret;

	if ((ret = director_cmd_is_seen_full(conn, &args, &seq, &dir_host)) < 0)
		return FALSE;

	if (str_array_length(args) != 1 ||
	    str_to_uint(args[0], &username_hash) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}

	if (ret > 0) {
		i_assert(dir_host != NULL);
		e_debug(conn->event,
			"User %u - ignoring already seen USER-KILLED-EVERYWHERE "
			"with seq=%u <= %s.last_seq=%u", username_hash,
			seq, dir_host->name, dir_host->last_seq);
		return TRUE;
	}

	director_user_killed_everywhere(conn->dir, conn->host,
					dir_host, username_hash);
	return TRUE;
}

static bool director_handshake_cmd_done(struct director_connection *conn)
{
	struct director *dir = conn->dir;
	int handshake_msecs = timeval_diff_msecs(&ioloop_timeval, &conn->connected_time);
	string_t *str;

	if (conn->users_unsorted && conn->user_iter == NULL) {
		/* we sent our user list before receiving remote's */
		conn->users_unsorted = FALSE;
		mail_hosts_sort_users(conn->dir->mail_hosts);
	}

	str = t_str_new(128);
	str_printfa(str, "Handshake finished in %u.%03u secs (",
		    handshake_msecs/1000, handshake_msecs%1000);
	director_connection_append_stats(conn, str);
	str_append_c(str, ')');
	if (handshake_msecs >= DIRECTOR_HANDSHAKE_WARN_SECS*1000)
		e_warning(conn->event, "%s", str_c(str));
	else
		e_info(conn->event, "%s", str_c(str));

	/* the host is up now, make sure we can connect to it immediately
	   if needed */
	conn->host->last_network_failure = 0;

	conn->handshake_received = TRUE;
	if (conn->in) {
		/* handshaked to left side. tell it we've received the
		   whole handshake. */
		director_connection_send(conn, "DONE\n");

		/* tell the "right" director about the "left" one */
		director_update_send(dir, director_connection_get_host(conn),
			t_strdup_printf("DIRECTOR\t%s\t%u\n",
					conn->host->ip_str,
					conn->host->port));
		/* this is our "left" side. */
		return director_connection_assign_left(conn);
	} else {
		/* handshaked to "right" side. */
		return director_connection_assign_right(conn);
	}
}

static int
director_handshake_cmd_options(struct director_connection *conn,
			       const char *const *args)
{
	bool consistent_hashing = FALSE;
	unsigned int i;

	for (i = 0; args[i] != NULL; i++) {
		if (strcmp(args[i], DIRECTOR_OPT_CONSISTENT_HASHING) == 0)
			consistent_hashing = TRUE;
	}
	if (!consistent_hashing) {
		e_error(conn->event, "director_consistent_hashing settings "
			"differ between directors. Set "
			"director_consistent_hashing=yes on old directors");
		return -1;
	}
	return 1;
}

static int
director_connection_handle_handshake(struct director_connection *conn,
				     const char *cmd, const char *const *args)
{
	unsigned int major_version;

	/* both incoming and outgoing connections get VERSION and ME */
	if (strcmp(cmd, "VERSION") == 0 && str_array_length(args) >= 3) {
		if (strcmp(args[0], DIRECTOR_VERSION_NAME) != 0) {
			e_error(conn->event, "Wrong protocol in socket "
				"(%s vs %s)",
				args[0], DIRECTOR_VERSION_NAME);
			return -1;
		} else if (str_to_uint(args[1], &major_version) < 0 ||
			str_to_uint(args[2], &conn->minor_version) < 0) {
			e_error(conn->event, "Invalid protocol version: "
				"%s.%s", args[1], args[2]);
			return -1;
		} else if (major_version != DIRECTOR_VERSION_MAJOR) {
			e_error(conn->event, "Incompatible protocol version: "
				"%u vs %u", major_version,
				DIRECTOR_VERSION_MAJOR);
			return -1;
		}
		if (conn->minor_version < DIRECTOR_VERSION_TAGS_V2 &&
		    mail_hosts_have_tags(conn->dir->mail_hosts)) {
			e_error(conn->event, "Director version supports incompatible tags");
			return -1;
		}
		conn->version_received = TRUE;
		director_finish_sending_handshake(conn);
		return 1;
	}
	if (!conn->version_received) {
		director_cmd_error(conn, "Incompatible protocol");
		return -1;
	}

	if (strcmp(cmd, "ME") == 0)
		return director_cmd_me(conn, args) ? 1 : -1;
	if (!conn->me_received) {
		director_cmd_error(conn, "Expecting ME command first");
		return -1;
	}

	/* incoming connections get a HOST list */
	if (conn->handshake_sending_hosts) {
		if (strcmp(cmd, "HOST") == 0)
			return director_cmd_host_handshake(conn, args) ? 1 : -1;
		if (strcmp(cmd, "HOST-HAND-END") == 0) {
			conn->ignore_host_events = FALSE;
			conn->handshake_sending_hosts = FALSE;
			return 1;
		}
		director_cmd_error(conn, "Unexpected command during host list");
		return -1;
	}
	if (strcmp(cmd, "OPTIONS") == 0)
		return director_handshake_cmd_options(conn, args);
	if (strcmp(cmd, "HOST-HAND-START") == 0) {
		if (!conn->in) {
			director_cmd_error(conn,
				"Host list is only for incoming connections");
			return -1;
		}
		return director_cmd_host_hand_start(conn, args) ? 1 : -1;
	}

	if (conn->in &&
	    (strcmp(cmd, "U") == 0 ||
	     (strcmp(cmd, "USER") == 0 &&
	      CMD_IS_USER_HANDSHAKE(conn->minor_version, args))))
		return director_handshake_cmd_user(conn, args) ? 1 : -1;

	/* both get DONE */
	if (strcmp(cmd, "DONE") == 0)
		return director_handshake_cmd_done(conn) ? 1 : -1;
	return 0;
}

static bool
director_connection_sync_host(struct director_connection *conn,
			      struct director_host *host,
			      uint32_t seq, unsigned int minor_version,
			      unsigned int timestamp, unsigned int hosts_hash)
{
	struct director *dir = conn->dir;

	if (minor_version > DIRECTOR_VERSION_MINOR) {
		/* we're not up to date */
		minor_version = DIRECTOR_VERSION_MINOR;
	}

	if (host->self) {
		if (dir->sync_seq != seq) {
			/* stale SYNC event */
			return FALSE;
		}
		/* sync_seq increases when we get disconnected, so we must be
		   successfully connected to both directions */
		i_assert(dir->left != NULL && dir->right != NULL);

		if (hosts_hash != 0 &&
		    hosts_hash != mail_hosts_hash(conn->dir->mail_hosts)) {
			e_error(conn->event, "Hosts unexpectedly changed during SYNC reply - resending"
				"(seq=%u, old hosts_hash=%u, new hosts_hash=%u)",
				seq, hosts_hash,
				mail_hosts_hash(dir->mail_hosts));
			(void)director_resend_sync(dir);
			return FALSE;
		}

		dir->ring_min_version = minor_version;
		if (!dir->ring_handshaked) {
			/* the ring is handshaked */
			director_set_ring_handshaked(dir);
		} else if (dir->ring_synced) {
			/* duplicate SYNC (which was sent just in case the
			   previous one got lost) */
		} else {
			e_debug(conn->event, "Ring is synced (%s sent seq=%u, hosts_hash=%u)",
				conn->name, seq,
				mail_hosts_hash(dir->mail_hosts));
			int sync_msecs =
				timeval_diff_msecs(&ioloop_timeval, &dir->last_sync_start_time);
			if (sync_msecs >= 0)
				dir->last_sync_msecs = sync_msecs;
			director_set_ring_synced(dir);
		}
	} else {
		if (seq < host->last_sync_seq &&
		    timestamp < host->last_sync_timestamp +
		    DIRECTOR_SYNC_STALE_TIMESTAMP_RESET_SECS) {
			/* stale SYNC event */
			e_debug(conn->event, "Ignore stale SYNC event for %s "
				"(seq %u < %u, timestamp=%u)",
				host->name, seq, host->last_sync_seq,
				timestamp);
			return FALSE;
		} else if (seq < host->last_sync_seq) {
			e_warning(conn->event,
				  "Last SYNC seq for %s appears to be stale, resetting "
				  "(seq=%u, timestamp=%u -> seq=%u, timestamp=%u)",
				  host->name, host->last_sync_seq,
				  host->last_sync_timestamp, seq, timestamp);
			host->last_sync_seq = seq;
			host->last_sync_timestamp = timestamp;
			host->last_sync_seq_counter = 1;
		} else if (seq > host->last_sync_seq ||
			   timestamp > host->last_sync_timestamp) {
			host->last_sync_seq = seq;
			host->last_sync_timestamp = timestamp;
			host->last_sync_seq_counter = 1;
			e_debug(conn->event, "Update SYNC for %s "
				"(seq=%u, timestamp=%u -> seq=%u, timestamp=%u)",
				host->name, host->last_sync_seq,
				host->last_sync_timestamp, seq, timestamp);
		} else if (++host->last_sync_seq_counter >
			   DIRECTOR_MAX_SYNC_SEQ_DUPLICATES) {
			/* we've received this too many times already */
			e_debug(conn->event, "Ignore duplicate #%u SYNC event for %s "
				"(seq=%u, timestamp %u <= %u)",
				host->last_sync_seq_counter, host->name, seq,
				timestamp, host->last_sync_timestamp);
			return FALSE;
		}

		if (hosts_hash != 0 &&
		    hosts_hash != mail_hosts_hash(conn->dir->mail_hosts)) {
			if (host->desynced_hosts_hash != hosts_hash) {
				e_debug(conn->event, "Ignore director %s stale SYNC request whose hosts don't match us "
					"(seq=%u, remote hosts_hash=%u, my hosts_hash=%u)",
					host->ip_str, seq, hosts_hash,
					mail_hosts_hash(dir->mail_hosts));
				host->desynced_hosts_hash = hosts_hash;
				return FALSE;
			}
			/* we'll get here only if we received a SYNC twice
			   with the same wrong hosts_hash. FIXME: this gets
			   triggered unnecessarily sometimes if hosts are
			   changing rapidly. */
			e_error(conn->event, "Director %s SYNC request hosts don't match us - resending hosts "
				"(seq=%u, remote hosts_hash=%u, my hosts_hash=%u)",
				host->ip_str, seq,
				hosts_hash, mail_hosts_hash(dir->mail_hosts));
			director_resend_hosts(dir);
			return FALSE;
		}
		host->desynced_hosts_hash = 0;
		if (dir->right != NULL) {
			/* forward it to the connection on right */
			director_sync_send(dir, host, seq, minor_version,
					   timestamp, hosts_hash);
		} else {
			e_debug(conn->event, "We have no right connection - "
				"delay replying to SYNC until finished");
			host->delayed_sync_seq = seq;
			host->delayed_sync_minor_version = minor_version;
			host->delayed_sync_timestamp = timestamp;
			host->delayed_sync_hosts_hash = hosts_hash;
		}
	}
	return TRUE;
}

static bool director_connection_sync(struct director_connection *conn,
				     const char *const *args)
{
	struct director *dir = conn->dir;
	struct director_host *host;
	struct ip_addr ip;
	in_port_t port;
	unsigned int arg_count, seq, minor_version = 0, timestamp = ioloop_time;
	unsigned int hosts_hash = 0;

	arg_count = str_array_length(args);
	if (arg_count < 3 ||
	    !director_args_parse_ip_port(conn, args, &ip, &port) ||
	    str_to_uint(args[2], &seq) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}
	if (arg_count >= 4 && str_to_uint(args[3], &minor_version) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}
	if (arg_count >= 5 && str_to_uint(args[4], &timestamp) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}
	if (arg_count >= 6 && str_to_uint(args[5], &hosts_hash) < 0) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}

	/* find the originating director. if we don't see it, it was already
	   removed and we can ignore this sync. */
	host = director_host_lookup(dir, &ip, port);
	if (host != NULL) {
		if (!director_connection_sync_host(conn, host, seq,
						   minor_version, timestamp,
						   hosts_hash))
			return TRUE;
	}

	/* If directors got disconnected while we were waiting a SYNC reply,
	   it might have gotten lost. If we've received a DIRECTOR update since
	   the last time we sent a SYNC, retry sending it here to make sure
	   it doesn't get stuck. We don't want to do this too eagerly because
	   it may trigger desynced_hosts_hash != hosts_hash mismatch, which
	   causes unnecessary error logging and hosts-resending. */
	if ((host == NULL || !host->self) &&
	    dir->last_sync_sent_ring_change_counter != dir->ring_change_counter &&
	    (time_t)dir->self_host->last_sync_timestamp != ioloop_time)
		(void)director_resend_sync(dir);
	return TRUE;
}

static void director_disconnect_timeout(struct director_connection *conn)
{
	director_connection_deinit(&conn, "CONNECT requested");
}

static void
director_reconnect_after_wrong_connect_timeout(struct director_connection *conn)
{
	struct director *dir = conn->dir;

	director_connection_deinit(&conn, "Wrong CONNECT requested");
	if (dir->right == NULL)
		director_connect(dir, "Reconnecting after wrong CONNECT request");
}

static void
director_reconnect_after_wrong_connect(struct director_connection *conn)
{
	if (conn->to_disconnect != NULL)
		return;
	conn->to_disconnect =
		timeout_add_short(DIRECTOR_RECONNECT_AFTER_WRONG_CONNECT_MSECS,
				  director_reconnect_after_wrong_connect_timeout, conn);
}

static bool director_cmd_connect(struct director_connection *conn,
				 const char *const *args)
{
	struct director *dir = conn->dir;
	struct director_host *host;
	struct ip_addr ip;
	in_port_t port;
	const char *right_state;

	if (str_array_length(args) != 2 ||
	    !director_args_parse_ip_port(conn, args, &ip, &port)) {
		director_cmd_error(conn, "Invalid parameters");
		return FALSE;
	}

	host = director_host_get(conn->dir, &ip, port);

	/* remote suggests us to connect elsewhere */
	if (dir->right != NULL &&
	    director_host_cmp_to_self(host, dir->right->host,
				      dir->self_host) <= 0) {
		/* the old connection is the correct one */
		e_debug(conn->event, "Ignoring CONNECT request to %s (current right is %s)",
			host->name, dir->right->name);
		director_reconnect_after_wrong_connect(conn);
		return TRUE;
	}
	if (host->removed) {
		e_debug(conn->event, "Ignoring CONNECT request to %s (director is removed)",
			host->name);
		director_reconnect_after_wrong_connect(conn);
		return TRUE;
	}

	/* reset failure timestamp so we'll actually try to connect there. */
	host->last_network_failure = 0;
	/* reset removed-flag, so we don't crash */
	host->removed = FALSE;

	if (dir->right == NULL) {
		right_state = "initializing right";
	} else {
		right_state = t_strdup_printf("replacing current right %s",
					      dir->right->name);
		/* disconnect from right side immediately - it's not accepting
		   any further commands from us. */
		if (conn->dir->right != conn)
			director_connection_deinit(&conn->dir->right, "CONNECT requested");
		else if (conn->to_disconnect == NULL) {
			conn->to_disconnect =
				timeout_add_short(0, director_disconnect_timeout, conn);
		}
	}

	/* connect here */
	(void)director_connect_host(dir, host, t_strdup_printf(
		"Received CONNECT request from %s - %s",
		conn->name, right_state));
	return TRUE;
}

static void director_disconnect_wrong_lefts(struct director *dir)
{
	struct director_connection *conn;

	array_foreach_elem(&dir->connections, conn) {
		if (conn->in && conn != dir->left && conn->me_received &&
		    conn->to_disconnect == NULL &&
		    director_host_cmp_to_self(dir->left->host, conn->host,
					      dir->self_host) < 0)
			director_connection_send_connect(conn, dir->left->host);
	}
}

static bool director_cmd_ping(struct director_connection *conn,
			      const char *const *args)
{
	time_t sent_time;
	uintmax_t send_buffer_size;

	if (str_array_length(args) >= 2 &&
	    str_to_time(args[0], &sent_time) == 0 &&
	    str_to_uintmax(args[1], &send_buffer_size) == 0) {
		int diff_secs = ioloop_time - sent_time;
		if (diff_secs*1000+500 > DIRECTOR_CONNECTION_PINGPONG_WARN_MSECS) {
			e_warning(conn->event,
				  "PING response took %d secs to receive "
				  "(send buffer was %ju bytes)",
				  diff_secs, send_buffer_size);
		}
	}
	director_connection_send(conn,
		t_strdup_printf("PONG\t%"PRIdTIME_T"\t%zu\n",
		ioloop_time, o_stream_get_buffer_used_size(conn->output)));
	return TRUE;
}

static void
director_ping_append_extra(struct director_connection *conn, string_t *str,
			   time_t pong_sent_time,
			   uintmax_t pong_send_buffer_size)
{
	struct rusage usage;

	str_printfa(str, "buffer size at PING was %zu bytes", conn->ping_sent_buffer_size);
	if (pong_sent_time != 0) {
		str_printfa(str, ", remote sent it %"PRIdTIME_T" secs ago",
			    ioloop_time - pong_sent_time);
	}
	if (pong_send_buffer_size != (uintmax_t)-1) {
		str_printfa(str, ", remote buffer size at PONG was %ju bytes",
			    pong_send_buffer_size);
	}
	if (conn->ping_sent_user_cpu.tv_sec != (time_t)-1 &&
	    getrusage(RUSAGE_SELF, &usage) == 0) {
		int diff = timeval_diff_msecs(&usage.ru_utime,
					      &conn->ping_sent_user_cpu);
		str_printfa(str, ", %u.%03u CPU secs since PING was sent",
			    diff/1000, diff%1000);
	}
	str_printfa(str, ", %"PRIuUOFF_T" bytes input",
		    conn->input->v_offset - conn->ping_sent_input_offset);
	str_printfa(str, ", %"PRIuUOFF_T" bytes output",
		    conn->output->offset - conn->ping_sent_output_offset);
}

static bool director_cmd_pong(struct director_connection *conn,
			      const char *const *args)
{
	time_t sent_time;
	uintmax_t send_buffer_size;

	if (!conn->ping_waiting)
		return TRUE;
	conn->ping_waiting = FALSE;
	timeout_remove(&conn->to_pong);

	if (str_array_length(args) < 2 ||
	    str_to_time(args[0], &sent_time) < 0 ||
	    str_to_uintmax(args[1], &send_buffer_size) < 0) {
		sent_time = 0;
		send_buffer_size = (uintmax_t)-1;
	}

	int ping_msecs = timeval_diff_msecs(&ioloop_timeval, &conn->ping_sent_time);
	if (ping_msecs >= 0) {
		if (ping_msecs > DIRECTOR_CONNECTION_PINGPONG_WARN_MSECS) {
			string_t *extra = t_str_new(128);
			director_ping_append_extra(conn, extra, sent_time, send_buffer_size);
			e_warning(conn->event,
				  "PONG response took %u.%03u secs (%s)",
				  ping_msecs/1000, ping_msecs%1000,
				  str_c(extra));
		}
		conn->last_ping_msecs = ping_msecs;
	}

	if (conn->verifying_left) {
		conn->verifying_left = FALSE;
		if (conn == conn->dir->left) {
			/* our left side is functional. tell all the wrong
			   incoming connections to connect to it instead. */
			director_disconnect_wrong_lefts(conn->dir);
		}
	}

	director_connection_set_ping_timeout(conn);
	return TRUE;
}

static bool
director_connection_handle_cmd(struct director_connection *conn,
			       const char *cmd, const char *const *args)
{
	int ret;

	if (!conn->handshake_received) {
		ret = director_connection_handle_handshake(conn, cmd, args);
		if (ret > 0)
			return TRUE;
		if (ret < 0) {
			/* invalid commands during handshake,
			   we probably don't want to reconnect here */
			return FALSE;
		}
		/* allow also other commands during handshake */
	}

	if (strcmp(cmd, "PING") == 0)
		return director_cmd_ping(conn, args);
	if (strcmp(cmd, "PONG") == 0)
		return director_cmd_pong(conn, args);
	if (strcmp(cmd, "USER") == 0)
		return director_cmd_user(conn, args);
	if (strcmp(cmd, "USER-WEAK") == 0)
		return director_cmd_user_weak(conn, args);
	if (strcmp(cmd, "HOST") == 0)
		return director_cmd_host(conn, args);
	if (strcmp(cmd, "HOST-REMOVE") == 0)
		return director_cmd_host_remove(conn, args);
	if (strcmp(cmd, "HOST-FLUSH") == 0)
		return director_cmd_host_flush(conn, args);
	if (strcmp(cmd, "USER-MOVE") == 0)
		return director_cmd_user_move(conn, args);
	if (strcmp(cmd, "USER-KICK") == 0)
		return director_cmd_user_kick(conn, args);
	if (strcmp(cmd, "USER-KICK-ALT") == 0)
		return director_cmd_user_kick_alt(conn, args);
	if (strcmp(cmd, "USER-KICK-HASH") == 0)
		return director_cmd_user_kick_hash(conn, args);
	if (strcmp(cmd, "USER-KILLED") == 0)
		return director_cmd_user_killed(conn, args);
	if (strcmp(cmd, "USER-KILLED-EVERYWHERE") == 0)
		return director_cmd_user_killed_everywhere(conn, args);
	if (strcmp(cmd, "DIRECTOR") == 0)
		return director_cmd_director(conn, args);
	if (strcmp(cmd, "DIRECTOR-REMOVE") == 0)
		return director_cmd_director_remove(conn, args);
	if (strcmp(cmd, "SYNC") == 0)
		return director_connection_sync(conn, args);
	if (strcmp(cmd, "CONNECT") == 0)
		return director_cmd_connect(conn, args);
	if (strcmp(cmd, "QUIT") == 0) {
		e_warning(conn->event,
			  "Director %s disconnected us with reason: %s",
			  conn->name, t_strarray_join(args, " "));
		return FALSE;
	}

	director_cmd_error(conn, "Unknown command %s", cmd);
	return FALSE;
}

static bool
director_connection_handle_line(struct director_connection *conn,
				char *line)
{
	const char *cmd, *const *args;
	bool ret;

	e_debug(conn->event, "input: %s", line);

	args = t_strsplit_tabescaped_inplace(line);
	cmd = args[0];
	if (cmd == NULL) {
		director_cmd_error(conn, "Received empty line");
		return FALSE;
	}

	conn->cur_cmd = cmd;
	conn->cur_args = args;
	ret = director_connection_handle_cmd(conn, cmd, args+1);
	conn->cur_cmd = NULL;
	conn->cur_args = NULL;
	return ret;
}

static void
director_connection_log_disconnect(struct director_connection *conn, int err,
				   const char *errstr)
{
	string_t *str = t_str_new(128);

	i_assert(conn->connected);

	if (conn->connect_request_to != NULL) {
		e_warning(conn->event,
			  "Director %s tried to connect to us, "
			  "should use %s instead",
			  conn->name, conn->connect_request_to->name);
		return;
	}

	str_printfa(str, "Director %s disconnected: ", conn->name);
	str_append(str, "Connection closed");
	if (err != 0 && err != EPIPE) {
		errno = err;
		if (errstr[0] == '\0')
			str_printfa(str, ": %m");
		else
			str_printfa(str, ": %s", errstr);
	}

	str_append(str, " (");
	director_connection_append_stats(conn, str);

	if (!conn->me_received)
		str_append(str, ", handshake ME not received");
	else if (!conn->handshake_received)
		str_append(str, ", handshake DONE not received");
	if (conn->synced)
		str_append(str, ", synced");
	str_append_c(str, ')');
	e_error(conn->event, "%s", str_c(str));
}

static void director_connection_input(struct director_connection *conn)
{
	struct director *dir = conn->dir;
	char *line;
	uoff_t prev_offset;
	bool ret;

	switch (i_stream_read(conn->input)) {
	case 0:
		return;
	case -1:
		/* disconnected */
		director_connection_log_disconnect(conn, conn->input->stream_errno,
						   i_stream_get_error(conn->input));
		director_connection_disconnected(&conn, i_stream_get_error(conn->input));
		return;
	case -2:
		/* buffer full */
		director_cmd_error(conn, "Director sent us more than %d bytes",
				   MAX_INBUF_SIZE);
		director_connection_reconnect(&conn, "Too long input line");
		return;
	}

	if (conn->to_disconnect != NULL) {
		/* just read everything the remote sends, and wait for it
		   to disconnect. we mainly just want the remote to read the
		   CONNECT we sent it. */
		i_stream_skip(conn->input, i_stream_get_data_size(conn->input));
		return;
	}
	conn->last_input = ioloop_timeval;
	conn->refcount++;

	director_sync_freeze(dir);
	prev_offset = conn->input->v_offset;
	while ((line = i_stream_next_line(conn->input)) != NULL) {
		dir->ring_traffic_input += conn->input->v_offset - prev_offset;
		prev_offset = conn->input->v_offset;

		T_BEGIN {
			ret = director_connection_handle_line(conn, line);
		} T_END;

		if (!ret) {
			if (!director_connection_unref(conn))
				break;
			director_connection_reconnect(&conn, t_strdup_printf(
				"Invalid input: %s", line));
			break;
		}
	}
	director_sync_thaw(dir);
	if (conn != NULL) {
		if (director_connection_unref(conn))
			timeout_reset(conn->to_ping);
	}
}

static void director_connection_send_directors(struct director_connection *conn)
{
	struct director_host *host;
	string_t *str = t_str_new(64);

	array_foreach_elem(&conn->dir->dir_hosts, host) {
		if (host->removed)
			continue;

		str_truncate(str, 0);
		str_printfa(str, "DIRECTOR\t%s\t%u\n",
			    host->ip_str, host->port);
		director_connection_send(conn, str_c(str));
	}
}

static void
director_connection_send_hosts(struct director_connection *conn)
{
	struct mail_host *host;
	bool send_updowns;
	string_t *str = t_str_new(128);

	i_assert(conn->version_received);

	send_updowns = conn->minor_version >= DIRECTOR_VERSION_UPDOWN;

	str_printfa(str, "HOST-HAND-START\t%u\n",
		    conn->dir->ring_handshaked ? 1 : 0);
	array_foreach_elem(mail_hosts_get(conn->dir->mail_hosts), host) {
		const char *host_tag = mail_host_get_tag(host);

		str_printfa(str, "HOST\t%s\t%u",
			    host->ip_str, host->vhost_count);
		if (host_tag[0] != '\0' || send_updowns) {
			str_append_c(str, '\t');
			str_append_tabescaped(str, host_tag);
		}
		if (send_updowns) {
			str_printfa(str, "\t%c%ld\t", host->down ? 'D' : 'U',
				    (long)host->last_updown_change);
			if (host->hostname != NULL)
				str_append_tabescaped(str, host->hostname);
		}
		str_append_c(str, '\n');
		director_connection_send(conn, str_c(str));
		str_truncate(str, 0);
	}
	str_printfa(str, "HOST-HAND-END\t%u\n",
		    conn->dir->ring_handshaked ? 1 : 0);
	director_connection_send(conn, str_c(str));
}

static int director_connection_send_done(struct director_connection *conn)
{
	i_assert(conn->version_received);

	if (conn->minor_version >= DIRECTOR_VERSION_OPTIONS) {
		director_connection_send(conn,
			"OPTIONS\t"DIRECTOR_OPT_CONSISTENT_HASHING"\n");
	} else {
		e_error(conn->event, "Director version is too old for supporting director_consistent_hashing=yes");
		return -1;
	}
	director_connection_send(conn, "DONE\n");
	return 0;
}

static int director_connection_send_users(struct director_connection *conn)
{
	struct user *user;
	string_t *str = t_str_new(128);
	char dec_buf[MAX_INT_STRLEN];
	unsigned int sent_count = 0;
	int ret;

	i_assert(conn->version_received);

	/* with new versions use "U" for sending the handshake users, because
	   otherwise their parameters may look identical and can't be
	   distinguished. */
	if (director_connection_get_minor_version(conn) >= DIRECTOR_VERSION_HANDSHAKE_U_CMD)
		str_append(str, "U\t");
	else
		str_append(str, "USER\t");
	size_t cmd_prefix_len = str_len(str);
	while ((user = director_iterate_users_next(conn->user_iter)) != NULL) {
		str_truncate(str, cmd_prefix_len);
		str_append(str, dec2str_buf(dec_buf, user->username_hash));
		str_append_c(str, '\t');
		str_append(str, user->host->ip_str);
		str_append_c(str, '\t');
		str_append(str, dec2str_buf(dec_buf, user->timestamp));
		if (user->weak)
			str_append(str, "\tw");
		str_append_c(str, '\n');

		conn->handshake_users_sent++;
		director_connection_send(conn, str_c(str));
		if (++sent_count >= DIRECTOR_HANDSHAKE_MAX_USERS_SENT_PER_FLUSH) {
			/* Don't send too much at once to avoid hangs */
			timeout_reset(conn->to_ping);
			return 0;
		}

		if (o_stream_get_buffer_used_size(conn->output) >= OUTBUF_FLUSH_THRESHOLD) {
			if ((ret = o_stream_flush(conn->output)) <= 0) {
				/* continue later */
				timeout_reset(conn->to_ping);
				return ret;
			}
		}
	}
	director_iterate_users_deinit(&conn->user_iter);
	if (director_connection_send_done(conn) < 0)
		return -1;

	if (conn->users_unsorted && conn->handshake_received) {
		/* we received remote's list of users before sending ours */
		conn->users_unsorted = FALSE;
		mail_hosts_sort_users(conn->dir->mail_hosts);
	}

	ret = o_stream_flush(conn->output);
	timeout_reset(conn->to_ping);
	return ret;
}

static int director_connection_output(struct director_connection *conn)
{
	int ret;

	conn->last_output = ioloop_timeval;
	if (conn->user_iter != NULL) {
		/* still handshaking USER list */
		ret = director_connection_send_users(conn);
		if (ret < 0) {
			director_connection_log_disconnect(conn, conn->output->stream_errno,
				o_stream_get_error(conn->output));
			director_connection_disconnected(&conn,
				o_stream_get_error(conn->output));
		} else {
			o_stream_set_flush_pending(conn->output, TRUE);
		}
		return ret;
	}
	return o_stream_flush(conn->output);
}

static struct director_connection *
director_connection_init_common(struct director *dir, int fd)
{
	struct director_connection *conn;

	conn = i_new(struct director_connection, 1);
	conn->refcount = 1;
	conn->created = ioloop_timeval;
	conn->fd = fd;
	conn->dir = dir;
	conn->event = event_create(dir->event);
	conn->input = i_stream_create_fd(conn->fd, MAX_INBUF_SIZE);
	conn->output = o_stream_create_fd(conn->fd, dir->set->director_output_buffer_size);
	o_stream_set_no_error_handling(conn->output, TRUE);
	array_push_back(&dir->connections, &conn);
	return conn;
}

static void director_connection_send_handshake(struct director_connection *conn)
{
	director_connection_send(conn, t_strdup_printf(
		"VERSION\t"DIRECTOR_VERSION_NAME"\t%u\t%u\n"
		"ME\t%s\t%u\t%lld\n",
		DIRECTOR_VERSION_MAJOR, DIRECTOR_VERSION_MINOR,
		net_ip2addr(&conn->dir->self_ip), conn->dir->self_port,
		(long long)time(NULL)));
}

static void director_connection_set_connected(struct director_connection *conn)
{
	struct rusage usage;

	conn->connected = TRUE;
	conn->connected_time = ioloop_timeval;

	if (getrusage(RUSAGE_SELF, &usage) == 0) {
		conn->connected_user_cpu_set = TRUE;
		conn->connected_user_cpu = usage.ru_utime;
	}
}

struct director_connection *
director_connection_init_in(struct director *dir, int fd,
			    const struct ip_addr *ip)
{
	struct director_connection *conn;

	conn = director_connection_init_common(dir, fd);
	conn->in = TRUE;
	director_connection_set_connected(conn);
	director_connection_set_name(conn,
		t_strdup_printf("%s/in", net_ip2addr(ip)));
	conn->io = io_add(conn->fd, IO_READ, director_connection_input, conn);
	conn->to_ping = timeout_add(DIRECTOR_CONNECTION_ME_TIMEOUT_MSECS,
				    director_connection_init_timeout, conn);

	e_info(conn->event, "Incoming connection from director %s", conn->name);
	director_connection_send_handshake(conn);
	return conn;
}

static void director_connection_connected(struct director_connection *conn)
{
	int err;

	if ((err = net_geterror(conn->fd)) != 0) {
		e_error(conn->event, "connect() failed: %s", strerror(err));
		director_connection_disconnected(&conn, strerror(err));
		return;
	}
	director_connection_set_connected(conn);
	o_stream_set_flush_callback(conn->output,
				    director_connection_output, conn);

	io_remove(&conn->io);
	conn->io = io_add(conn->fd, IO_READ, director_connection_input, conn);

	timeout_remove(&conn->to_ping);
	conn->to_ping = timeout_add(DIRECTOR_CONNECTION_ME_TIMEOUT_MSECS,
				    director_connection_init_timeout, conn);

	o_stream_cork(conn->output);
	director_connection_send_handshake(conn);
	director_connection_send_directors(conn);
	o_stream_uncork(conn->output);
	/* send the rest of the handshake after we've received the remote's
	   version number */
}

static void director_finish_sending_handshake(struct director_connection *conn)
{
	if (
	    conn->in) {
		/* only outgoing connections send hosts & users */
		return;
	}
	o_stream_cork(conn->output);
	director_connection_send_hosts(conn);

	i_assert(conn->user_iter == NULL);
	/* Iterate only through users that aren't refreshed since the
	   iteration started. The refreshed users will already be sent as
	   regular USER updates, so they don't need to be sent again.

	   We especially don't want to send these users again, because
	   otherwise in a rapidly changing director we might never end up
	   sending all the users when they constantly keep being added to the
	   end of the list. (The iteration lists users in order from older to
	   newer.) */
	conn->user_iter = director_iterate_users_init(conn->dir, TRUE);

	if (director_connection_send_users(conn) == 0)
		o_stream_set_flush_pending(conn->output, TRUE);

	o_stream_uncork(conn->output);
}

struct director_connection *
director_connection_init_out(struct director *dir, int fd,
			     struct director_host *host)
{
	struct director_connection *conn;

	i_assert(!host->removed);

	/* make sure we don't keep old sequence values across restarts */
	director_host_restarted(host);

	conn = director_connection_init_common(dir, fd);
	director_connection_set_name(conn,
		t_strdup_printf("%s/out", host->name));
	conn->host = host;
	director_host_ref(host);
	conn->io = io_add(conn->fd, IO_WRITE,
			  director_connection_connected, conn);
	conn->to_ping = timeout_add(DIRECTOR_CONNECTION_CONNECT_TIMEOUT_MSECS,
				    director_connection_init_timeout, conn);
	return conn;
}

void director_connection_deinit(struct director_connection **_conn,
				const char *remote_reason)
{
	struct director_connection *const *conns, *conn = *_conn;
	struct director *dir = conn->dir;
	unsigned int i, count;

	*_conn = NULL;

	i_assert(conn->fd != -1);

	if (conn->host != NULL) {
		e_debug(conn->event, "Disconnecting from %s: %s",
			conn->host->name, remote_reason);
	}
	if (*remote_reason != '\0' &&
	    conn->minor_version >= DIRECTOR_VERSION_QUIT) {
		o_stream_nsend_str(conn->output, t_strdup_printf(
			"QUIT\t%s\n", remote_reason));
	}

	conns = array_get(&dir->connections, &count);
	for (i = 0; i < count; i++) {
		if (conns[i] == conn) {
			array_delete(&dir->connections, i, 1);
			break;
		}
	}
	i_assert(i < count);
	if (dir->left == conn) {
		dir->left = NULL;
		/* if there is already another handshaked incoming connection,
		   use it as the new "left" */
		director_assign_left(dir);
	}
	if (dir->right == conn)
		dir->right = NULL;

	if (conn->users_unsorted) {
		/* Users were received, but handshake didn't finish.
		   Finish sorting so the users won't stay in wrong order. */
		mail_hosts_sort_users(conn->dir->mail_hosts);
	}

	if (conn->connect_request_to != NULL) {
		director_host_unref(conn->connect_request_to);
		conn->connect_request_to = NULL;
	}
	if (conn->user_iter != NULL)
		director_iterate_users_deinit(&conn->user_iter);
	timeout_remove(&conn->to_disconnect);
	timeout_remove(&conn->to_pong);
	timeout_remove(&conn->to_ping);
	io_remove(&conn->io);
	i_stream_close(conn->input);
	o_stream_close(conn->output);
	i_close_fd(&conn->fd);

	if (conn->in)
		master_service_client_connection_destroyed(master_service);
	director_connection_unref(conn);

	if (dir->left == NULL || dir->right == NULL) {
		/* we aren't synced until we're again connected to a ring */
		dir->sync_seq++;
		director_set_ring_unsynced(dir);
	}
}

static bool director_connection_unref(struct director_connection *conn)
{
	i_assert(conn->refcount > 0);
	if (--conn->refcount > 0)
		return TRUE;

	if (conn->host != NULL)
		director_host_unref(conn->host);
	i_stream_unref(&conn->input);
	o_stream_unref(&conn->output);
	event_unref(&conn->event);
	i_free(conn->name);
	i_free(conn);
	return FALSE;
}

static void director_connection_disconnected(struct director_connection **_conn,
					     const char *reason)
{
	struct director_connection *conn = *_conn;
	struct director *dir = conn->dir;

	if ((conn->connected_time.tv_sec == 0 ||
	     conn->connected_time.tv_sec + DIRECTOR_SUCCESS_MIN_CONNECT_SECS > ioloop_time) &&
	    conn->host != NULL) {
		/* connection didn't exist for very long, assume it has a
		   network problem */
		conn->host->last_network_failure = ioloop_time;
	}

	director_connection_deinit(_conn, reason);
	if (dir->right == NULL)
		director_connect(dir, "Reconnecting after disconnection");
}

static void director_connection_reconnect(struct director_connection **_conn,
					  const char *reason)
{
	struct director_connection *conn = *_conn;
	struct director *dir = conn->dir;

	director_connection_deinit(_conn, reason);
	if (dir->right == NULL)
		director_connect(dir, "Reconnecting after error");
}

static void director_disconnect_write_error(struct director_connection *conn)
{
	struct director *dir = conn->dir;

	director_connection_deinit(&conn, "write failure");
	if (dir->right == NULL)
		director_connect(dir, "Reconnecting after write failure");
}

void director_connection_send(struct director_connection *conn,
			      const char *data)
{
	size_t len = strlen(data);
	off_t ret;

	if (conn->output->closed || !conn->connected)
		return;

	if (event_want_debug(conn->event)) T_BEGIN {
		const char *const *lines = t_strsplit(data, "\n");
		for (; lines[1] != NULL; lines++)
			e_debug(conn->event, "output: %s", *lines);
	} T_END;
	ret = o_stream_send(conn->output, data, len);
	if (ret != (off_t)len) {
		if (ret < 0) {
			director_connection_log_disconnect(conn,
				conn->output->stream_errno,
				t_strdup_printf("write() failed: %s",
					o_stream_get_error(conn->output)));
		} else {
			director_connection_log_disconnect(conn, EINVAL,
				t_strdup_printf("Output buffer full at %zu",
					o_stream_get_buffer_used_size(conn->output)));
		}
		o_stream_close(conn->output);
		/* closing the stream when output buffer is full doesn't cause
		   disconnection itself. */
		timeout_remove(&conn->to_disconnect);
		conn->to_disconnect =
			timeout_add_short(0, director_disconnect_write_error, conn);
	} else {
		conn->dir->ring_traffic_output += len;
		conn->last_output = ioloop_timeval;
		conn->peak_bytes_buffered =
			I_MAX(conn->peak_bytes_buffered,
			      o_stream_get_buffer_used_size(conn->output));
	}
}

static void
director_connection_ping_idle_timeout(struct director_connection *conn)
{
	string_t *str = t_str_new(128);
	int diff = timeval_diff_msecs(&ioloop_timeval, &conn->ping_sent_time);

	str_printfa(str, "Ping timed out in %u.%03u secs: ",
		    diff/1000, diff%1000);
	director_ping_append_extra(conn, str, 0, (uintmax_t)-1);
	director_connection_log_disconnect(conn, EINVAL, str_c(str));
	director_connection_disconnected(&conn, "Ping timeout");
}

static void director_connection_pong_timeout(struct director_connection *conn)
{
	int diff = timeval_diff_msecs(&ioloop_timeval, &conn->ping_sent_time);
	const char *errstr;

	errstr = t_strdup_printf(
		"PONG reply not received in %u.%03u secs, "
		"although other input keeps coming",
		diff/1000, diff%1000);
	director_connection_log_disconnect(conn, EINVAL, errstr);
	director_connection_disconnected(&conn, "Pong timeout");
}

void director_connection_ping(struct director_connection *conn)
{
	if (conn->ping_waiting)
		return;

	timeout_remove(&conn->to_ping);
	conn->to_ping = timeout_add(conn->dir->set->director_ping_idle_timeout*1000,
				    director_connection_ping_idle_timeout, conn);
	conn->to_pong = timeout_add(conn->dir->set->director_ping_max_timeout*1000,
				    director_connection_pong_timeout, conn);
	conn->ping_waiting = TRUE;
	conn->ping_sent_time = ioloop_timeval;
	conn->ping_sent_buffer_size = o_stream_get_buffer_used_size(conn->output);
	conn->ping_sent_input_offset = conn->input->v_offset;
	conn->ping_sent_output_offset = conn->output->offset;

	struct rusage usage;
	if (getrusage(RUSAGE_SELF, &usage) == 0)
		conn->ping_sent_user_cpu = usage.ru_utime;
	else
		conn->ping_sent_user_cpu.tv_sec = (time_t)-1;
	/* send it after getting the buffer size */
	director_connection_send(conn, t_strdup_printf(
		"PING\t%"PRIdTIME_T"\t%zu\n", ioloop_time,
		conn->ping_sent_buffer_size));
}

const char *director_connection_get_name(struct director_connection *conn)
{
	return conn->name;
}

struct director_host *
director_connection_get_host(struct director_connection *conn)
{
	return conn->host;
}

bool director_connection_is_handshaked(struct director_connection *conn)
{
	return conn->handshake_received;
}

bool director_connection_is_synced(struct director_connection *conn)
{
	return conn->synced;
}

bool director_connection_is_incoming(struct director_connection *conn)
{
	return conn->in;
}

unsigned int
director_connection_get_minor_version(struct director_connection *conn)
{
	return conn->minor_version;
}

void director_connection_cork(struct director_connection *conn)
{
	o_stream_cork(conn->output);
}

void director_connection_uncork(struct director_connection *conn)
{
	o_stream_uncork(conn->output);
}

void director_connection_set_synced(struct director_connection *conn,
				    bool synced)
{
	if (conn->synced == synced)
		return;
	conn->synced = synced;

	/* switch ping timeout, unless we're already waiting for PONG */
	if (conn->ping_waiting)
		return;

	director_connection_set_ping_timeout(conn);
}

void director_connection_get_status(struct director_connection *conn,
				    struct director_connection_status *status_r)
{
	i_zero(status_r);
	status_r->bytes_read = conn->input->v_offset;
	status_r->bytes_sent = conn->output->offset;
	status_r->bytes_buffered = o_stream_get_buffer_used_size(conn->output);
	status_r->peak_bytes_buffered = conn->peak_bytes_buffered;
	status_r->last_input = conn->last_input;
	status_r->last_output = conn->last_output;
	status_r->last_ping_msecs = conn->last_ping_msecs;
	status_r->handshake_users_sent = conn->handshake_users_sent;
	status_r->handshake_users_received = conn->handshake_users_received;
}