2892N/A

2892N/A#include "lib.h"

2892N/A#include "ioloop.h"

2892N/A#include "array.h"

6029N/A#include "str.h"

2521N/A#include "strescape.h"

2521N/A#include "log-throttle.h"

2521N/A#include "ipc-client.h"

2521N/A#include "program-client.h"

2521N/A#include "var-expand.h"

2521N/A#include "istream.h"

2521N/A#include "ostream.h"

2521N/A#include "iostream-temp.h"

2521N/A#include "user-directory.h"

2521N/A#include "mail-host.h"

2521N/A#include "director-host.h"

2521N/A#include "director-connection.h"

2521N/A#include "director.h"

2521N/A

5403N/A#define DIRECTOR_IPC_PROXY_PATH "ipc"

2521N/A#define DIRECTOR_DNS_SOCKET_PATH "dns-client"

2892N/A#define DIRECTOR_RECONNECT_RETRY_SECS 60

5403N/A#define DIRECTOR_RECONNECT_TIMEOUT_MSECS (30*1000)

2521N/A#define DIRECTOR_USER_MOVE_TIMEOUT_MSECS (30*1000)

2892N/A#define DIRECTOR_SYNC_TIMEOUT_MSECS (5*1000)

5403N/A#define DIRECTOR_RING_MIN_WAIT_SECS 20

2892N/A#define DIRECTOR_QUICK_RECONNECT_TIMEOUT_MSECS 1000

3524N/A#define DIRECTOR_DELAYED_DIR_REMOVE_MSECS (1000*30)

2892N/A

4973N/Abool director_debug;

5403N/A

2521N/Aconst char *user_kill_state_names[USER_KILL_STATE_DELAY+1] = {

2521N/A "none",

2521N/A "killing",

2521N/A "notify-received",

5403N/A "waiting-for-notify",

5403N/A "waiting-for-everyone",

2521N/A "flushing",

5403N/A "delay",

5403N/A};

5403N/A

5403N/Astatic struct log_throttle *user_move_throttle;

5403N/Astatic struct log_throttle *user_kill_fail_throttle;

2521N/A

5403N/Astatic const struct log_throttle_settings director_log_throttle_settings = {

2521N/A .throttle_at_max_per_interval = 100,

2521N/A .unthrottle_at_max_per_interval = 2,

2521N/A};

2521N/A

2521N/Astatic void

2521N/Adirector_user_kill_finish_delayed(struct director *dir, struct user *user,

2521N/A bool skip_delay);

2521N/A

5403N/Astatic bool director_is_self_ip_set(struct director *dir)

5403N/A{

5403N/A if (net_ip_compare(&dir->self_ip, &net_ip4_any))

5403N/A return FALSE;

2521N/A

5403N/A if (net_ip_compare(&dir->self_ip, &net_ip6_any))

2521N/A return FALSE;

2521N/A

2892N/A return TRUE;

2521N/A}

3998N/A

2521N/Astatic void director_find_self_ip(struct director *dir)

2521N/A{

2521N/A struct director_host *const *hosts;

2521N/A unsigned int i, count;

2521N/A

2521N/A hosts = array_get(&dir->dir_hosts, &count);

2892N/A for (i = 0; i < count; i++) {

2763N/A if (net_try_bind(&hosts[i]->ip) == 0) {

2763N/A dir->self_ip = hosts[i]->ip;

2763N/A return;

2763N/A }

2763N/A }

3998N/A i_fatal("director_servers doesn't list ourself");

3998N/A}

2763N/A

2763N/Avoid director_find_self(struct director *dir)

2763N/A{

3998N/A if (dir->self_host != NULL)

3998N/A return;

3998N/A

3998N/A if (!director_is_self_ip_set(dir))

3998N/A director_find_self_ip(dir);

3998N/A

3998N/A dir->self_host = director_host_lookup(dir, &dir->self_ip,

3998N/A dir->self_port);

3998N/A if (dir->self_host == NULL) {

2892N/A i_fatal("director_servers doesn't list ourself (%s:%u)",

2892N/A net_ip2addr(&dir->self_ip), dir->self_port);

2892N/A }

3998N/A dir->self_host->self = TRUE;

3998N/A}

3998N/A

3998N/Astatic unsigned int director_find_self_idx(struct director *dir)

3998N/A{

3998N/A struct director_host *const *hosts;

3998N/A unsigned int i, count;

3998N/A

3998N/A i_assert(dir->self_host != NULL);

3998N/A

3998N/A hosts = array_get(&dir->dir_hosts, &count);

3998N/A for (i = 0; i < count; i++) {

3998N/A if (hosts[i] == dir->self_host)

3998N/A return i;

3998N/A }

3998N/A i_unreached();

3998N/A}

2892N/A

5403N/Astatic bool

5403N/Adirector_has_outgoing_connection(struct director *dir,

5403N/A struct director_host *host)

5403N/A{

5403N/A struct director_connection *const *connp;

5403N/A

2892N/A array_foreach(&dir->connections, connp) {

2892N/A if (director_connection_get_host(*connp) == host &&

2892N/A !director_connection_is_incoming(*connp))

2892N/A return TRUE;

2521N/A }

2521N/A return FALSE;

3998N/A}

3998N/A

3998N/Aint director_connect_host(struct director *dir, struct director_host *host)

3998N/A{

2521N/A in_port_t port;

2521N/A int fd;

2521N/A

2892N/A if (director_has_outgoing_connection(dir, host))

5403N/A return 0;

5403N/A

5403N/A if (director_debug) {

5403N/A string_t *str = t_str_new(128);

2892N/A

3998N/A str_printfa(str, "Connecting to %s:%u (as %s",

2959N/A net_ip2addr(&host->ip), host->port,

3998N/A net_ip2addr(&dir->self_ip));

3998N/A if (host->last_network_failure > 0) {

5403N/A str_printfa(str, ", last network failure %ds ago",

2521N/A (int)(ioloop_time - host->last_network_failure));

2892N/A }

2892N/A if (host->last_protocol_failure > 0) {

5403N/A str_printfa(str, ", last protocol failure %ds ago",

2892N/A (int)(ioloop_time - host->last_protocol_failure));

2892N/A }

3998N/A dir_debug("%s", str_c(str));

3998N/A }

5403N/A port = dir->test_port != 0 ? dir->test_port : host->port;

5403N/A fd = net_connect_ip(&host->ip, port, &dir->self_ip);

3998N/A if (fd == -1) {

2521N/A host->last_network_failure = ioloop_time;

2892N/A i_error("connect(%s) failed: %m", host->name);

5403N/A return -1;

2892N/A }

5403N/A /* Reset timestamp so that director_connect() won't skip this host

3998N/A host->last_network_failure = 0;

2892N/A

2892N/A (void)director_connection_init_out(dir, fd, host);

2892N/A return 0;

2892N/A}

2892N/A

2892N/Astatic struct director_host *

2892N/Adirector_get_preferred_right_host(struct director *dir)

2892N/A{

2892N/A struct director_host *const *hosts, *host;

2892N/A unsigned int i, count, self_idx;

2892N/A

2892N/A hosts = array_get(&dir->dir_hosts, &count);

2892N/A if (count == 1) {

2892N/A /* self */

3524N/A return NULL;

3524N/A }

3524N/A

3524N/A self_idx = director_find_self_idx(dir);

3524N/A for (i = 0; i < count; i++) {

3524N/A host = hosts[(self_idx + i + 1) % count];

3524N/A if (!host->removed)

3524N/A return host;

2892N/A }

2892N/A /* self, with some removed hosts */

2892N/A return NULL;

2892N/A}

2892N/A

2892N/Astatic bool director_wait_for_others(struct director *dir)

2892N/A{

2892N/A struct director_host *const *hostp;

2892N/A

2892N/A /* don't assume we're alone until we've attempted to connect

2892N/A if (dir->ring_first_alone != 0 &&

2892N/A ioloop_time - dir->ring_first_alone > DIRECTOR_RING_MIN_WAIT_SECS)

3998N/A return FALSE;

3998N/A

3998N/A if (dir->ring_first_alone == 0)

3998N/A dir->ring_first_alone = ioloop_time;

2892N/A /* reset all failures and try again */

2892N/A array_foreach(&dir->dir_hosts, hostp) {

2892N/A (*hostp)->last_network_failure = 0;

2892N/A (*hostp)->last_protocol_failure = 0;

2892N/A }

3998N/A if (dir->to_reconnect != NULL)

3998N/A timeout_remove(&dir->to_reconnect);

5403N/A dir->to_reconnect = timeout_add(DIRECTOR_QUICK_RECONNECT_TIMEOUT_MSECS,

5403N/A director_connect, dir);

3998N/A return TRUE;

3998N/A}

3998N/A

3998N/Avoid director_connect(struct director *dir)

3998N/A{

3998N/A struct director_host *const *hosts;

3998N/A unsigned int i, count, self_idx;

3998N/A

3998N/A self_idx = director_find_self_idx(dir);

3998N/A

2892N/A /* try to connect to first working server on our right side.

5403N/A hosts = array_get(&dir->dir_hosts, &count);

5403N/A for (i = 1; i < count; i++) {

5403N/A unsigned int idx = (self_idx + i) % count;

2892N/A

2892N/A if (hosts[idx]->removed)

2892N/A continue;

2892N/A

2892N/A if (hosts[idx]->last_network_failure +

2892N/A DIRECTOR_RECONNECT_RETRY_SECS > ioloop_time) {

2892N/A /* connection failed recently, don't try retrying here */

2892N/A continue;

2892N/A }

3998N/A if (hosts[idx]->last_protocol_failure +

3998N/A DIRECTOR_PROTOCOL_FAILURE_RETRY_SECS > ioloop_time) {

3998N/A /* the director recently sent invalid protocol data,

2892N/A continue;

2892N/A }

2892N/A

2892N/A if (director_connect_host(dir, hosts[idx]) == 0) {

2892N/A /* success */

5403N/A return;

2892N/A }

2892N/A }

2892N/A

2892N/A if (count > 1 && director_wait_for_others(dir))

2892N/A return;

3998N/A

2892N/A /* we're the only one */

2892N/A if (count > 1) {

2892N/A i_warning("director: Couldn't connect to right side, "

2892N/A "we must be the only director left");

2892N/A }

5403N/A if (dir->left != NULL) {

2892N/A /* since we couldn't connect to it,

6029N/A i_warning("director: Assuming %s is dead, disconnecting",

6029N/A director_connection_get_name(dir->left));

2892N/A director_connection_deinit(&dir->left,

4973N/A "This connection is dead?");

2892N/A }

2892N/A dir->ring_min_version = DIRECTOR_VERSION_MINOR;

4973N/A if (!dir->ring_handshaked)

4973N/A director_set_ring_handshaked(dir);

4973N/A else

4973N/A director_set_ring_synced(dir);

4973N/A}

4973N/A

2892N/Avoid director_set_ring_handshaked(struct director *dir)

5403N/A{

3998N/A i_assert(!dir->ring_handshaked);

2892N/A

2892N/A if (dir->to_handshake_warning != NULL)

2892N/A timeout_remove(&dir->to_handshake_warning);

2892N/A if (dir->ring_handshake_warning_sent) {

2892N/A i_warning("Directors have been connected, "

2892N/A "continuing delayed requests");

2892N/A dir->ring_handshake_warning_sent = FALSE;

2892N/A }

2892N/A dir_debug("Director ring handshaked");

dir->ring_handshaked = TRUE;

director_set_ring_synced(dir);

}

static void director_reconnect_timeout(struct director *dir)

{

struct director_host *cur_host, *preferred_host =

director_get_preferred_right_host(dir);

cur_host = dir->right == NULL ? NULL :

director_connection_get_host(dir->right);

if (preferred_host == NULL) {

/* all directors have been removed, try again later */

} else if (cur_host != preferred_host)

(void)director_connect_host(dir, preferred_host);

else {

/* the connection hasn't finished sync yet.

}

}

void director_set_ring_synced(struct director *dir)

{

struct director_host *host;

i_assert(!dir->ring_synced);

i_assert((dir->left != NULL && dir->right != NULL) ||

(dir->left == NULL && dir->right == NULL));

if (dir->to_handshake_warning != NULL)

timeout_remove(&dir->to_handshake_warning);

if (dir->ring_handshake_warning_sent) {

i_warning("Ring is synced, continuing delayed requests "

"(syncing took %d secs, hosts_hash=%u)",

(int)(ioloop_time - dir->ring_last_sync_time),

mail_hosts_hash(dir->mail_hosts));

dir->ring_handshake_warning_sent = FALSE;

}

host = dir->right == NULL ? NULL :

director_connection_get_host(dir->right);

if (dir->to_reconnect != NULL)

timeout_remove(&dir->to_reconnect);

if (host != director_get_preferred_right_host(dir)) {

/* try to reconnect to preferred host later */

dir->to_reconnect =

timeout_add(DIRECTOR_RECONNECT_TIMEOUT_MSECS,

director_reconnect_timeout, dir);

}

if (dir->left != NULL)

director_connection_set_synced(dir->left, TRUE);

if (dir->right != NULL)

director_connection_set_synced(dir->right, TRUE);

if (dir->to_sync != NULL)

timeout_remove(&dir->to_sync);

dir->ring_synced = TRUE;

dir->ring_last_sync_time = ioloop_time;

mail_hosts_set_synced(dir->mail_hosts);

director_set_state_changed(dir);

}

void director_sync_send(struct director *dir, struct director_host *host,

uint32_t seq, unsigned int minor_version,

unsigned int timestamp, unsigned int hosts_hash)

{

string_t *str;

if (host == dir->self_host)

dir->last_sync_sent_ring_change_counter = dir->ring_change_counter;

str = t_str_new(128);

str_printfa(str, "SYNC\t%s\t%u\t%u",

net_ip2addr(&host->ip), host->port, seq);

if (minor_version > 0 &&

director_connection_get_minor_version(dir->right) > 0) {

/* only minor_version>0 supports extra parameters */

str_printfa(str, "\t%u\t%u\t%u", minor_version,

timestamp, hosts_hash);

}

str_append_c(str, '\n');

director_connection_send(dir->right, str_c(str));

/* ping our connections in case either of them are hanging.

if (dir->left != NULL)

director_connection_ping(dir->left);

director_connection_ping(dir->right);

}

bool director_resend_sync(struct director *dir)

{

if (!dir->ring_synced && dir->left != NULL && dir->right != NULL) {

/* send a new SYNC in case the previous one got dropped */

dir->self_host->last_sync_timestamp = ioloop_time;

director_sync_send(dir, dir->self_host, dir->sync_seq,

DIRECTOR_VERSION_MINOR, ioloop_time,

mail_hosts_hash(dir->mail_hosts));

if (dir->to_sync != NULL)

timeout_reset(dir->to_sync);

return TRUE;

}

return FALSE;

}

static void director_sync_timeout(struct director *dir)

{

i_assert(!dir->ring_synced);

if (director_resend_sync(dir))

i_error("Ring SYNC appears to have got lost, resending");

}

void director_set_ring_unsynced(struct director *dir)

{

if (dir->ring_synced) {

dir->ring_synced = FALSE;

dir->ring_last_sync_time = ioloop_time;

}

if (dir->to_sync == NULL) {

dir->to_sync = timeout_add(DIRECTOR_SYNC_TIMEOUT_MSECS,

director_sync_timeout, dir);

} else {

timeout_reset(dir->to_sync);

}

}

static void director_sync(struct director *dir)

{

/* we're synced again when we receive this SYNC back */

dir->sync_seq++;

if (dir->right == NULL && dir->left == NULL) {

/* we're alone. if we're already synced,

return;

}

director_set_ring_unsynced(dir);

if (dir->sync_frozen) {

dir->sync_pending = TRUE;

return;

}

if (dir->right == NULL) {

i_assert(!dir->ring_synced ||

(dir->left == NULL && dir->right == NULL));

return;

}

dir_debug("Ring is desynced (seq=%u, sending SYNC to %s)",

dir->sync_seq, dir->right == NULL ? "(nowhere)" :

director_connection_get_name(dir->right));

/* send PINGs to our connections more rapidly until we've synced again.

if (dir->left != NULL)

director_connection_set_synced(dir->left, FALSE);

director_connection_set_synced(dir->right, FALSE);

director_sync_send(dir, dir->self_host, dir->sync_seq,

DIRECTOR_VERSION_MINOR, ioloop_time,

mail_hosts_hash(dir->mail_hosts));

}

void director_sync_freeze(struct director *dir)

{

struct director_connection *const *connp;

i_assert(!dir->sync_frozen);

i_assert(!dir->sync_pending);

array_foreach(&dir->connections, connp)

director_connection_cork(*connp);

dir->sync_frozen = TRUE;

}

void director_sync_thaw(struct director *dir)

{

struct director_connection *const *connp;

i_assert(dir->sync_frozen);

dir->sync_frozen = FALSE;

if (dir->sync_pending) {

dir->sync_pending = FALSE;

director_sync(dir);

}

array_foreach(&dir->connections, connp)

director_connection_uncork(*connp);

}

void director_notify_ring_added(struct director_host *added_host,

struct director_host *src)

{

const char *cmd;

added_host->dir->ring_change_counter++;

cmd = t_strdup_printf("DIRECTOR\t%s\t%u\n",

net_ip2addr(&added_host->ip), added_host->port);

director_update_send(added_host->dir, src, cmd);

}

static void director_delayed_dir_remove_timeout(struct director *dir)

{

struct director_host *const *hosts, *host;

unsigned int i, count;

timeout_remove(&dir->to_remove_dirs);

hosts = array_get(&dir->dir_hosts, &count);

for (i = 0; i < count; ) {

if (hosts[i]->removed) {

host = hosts[i];

director_host_free(&host);

hosts = array_get(&dir->dir_hosts, &count);

} else {

i++;

}

}

}

void director_ring_remove(struct director_host *removed_host,

struct director_host *src)

{

struct director *dir = removed_host->dir;

struct director_connection *const *conns, *conn;

unsigned int i, count;

const char *cmd;

if (removed_host->self) {

/* others will just disconnect us */

return;

}

/* mark the host as removed and fully remove it later. this delay is

removed_host->removed = TRUE;

if (dir->to_remove_dirs == NULL) {

dir->to_remove_dirs =

timeout_add(DIRECTOR_DELAYED_DIR_REMOVE_MSECS,

director_delayed_dir_remove_timeout, dir);

}

/* disconnect any connections to the host */

conns = array_get(&dir->connections, &count);

for (i = 0; i < count; ) {

conn = conns[i];

if (director_connection_get_host(conn) != removed_host)

i++;

else {

director_connection_deinit(&conn, "Removing from ring");

conns = array_get(&dir->connections, &count);

}

}

if (dir->right == NULL)

director_connect(dir);

cmd = t_strdup_printf("DIRECTOR-REMOVE\t%s\t%u\n",

net_ip2addr(&removed_host->ip),

removed_host->port);

director_update_send_version(dir, src,

DIRECTOR_VERSION_RING_REMOVE, cmd);

}

static void

director_send_host(struct director *dir, struct director_host *src,

struct director_host *orig_src,

struct mail_host *host)

{

const char *host_tag = mail_host_get_tag(host);

string_t *str;

if (orig_src == NULL) {

orig_src = dir->self_host;

orig_src->last_seq++;

}

str = t_str_new(128);

str_printfa(str, "HOST\t%s\t%u\t%u\t%s\t%u",

net_ip2addr(&orig_src->ip), orig_src->port,

orig_src->last_seq,

net_ip2addr(&host->ip), host->vhost_count);

if (dir->ring_min_version >= DIRECTOR_VERSION_TAGS_V2) {

str_append_c(str, '\t');

str_append_tabescaped(str, host_tag);

} else if (host_tag[0] != '\0' &&

dir->ring_min_version < DIRECTOR_VERSION_TAGS_V2) {

if (dir->ring_min_version < DIRECTOR_VERSION_TAGS) {

i_error("Ring has directors that don't support tags - removing host %s with tag '%s'",

net_ip2addr(&host->ip), host_tag);

} else {

i_error("Ring has directors that support mixed versions of tags - removing host %s with tag '%s'",

net_ip2addr(&host->ip), host_tag);

}

director_remove_host(dir, NULL, NULL, host);

return;

}

if (dir->ring_min_version >= DIRECTOR_VERSION_UPDOWN) {

str_printfa(str, "\t%c%ld\t", host->down ? 'D' : 'U',

(long)host->last_updown_change);

/* add any further version checks here - these directors ignore

if (host->hostname != NULL)

str_append_tabescaped(str, host->hostname);

}

str_append_c(str, '\n');

director_update_send(dir, src, str_c(str));

}

void director_resend_hosts(struct director *dir)

{

struct mail_host *const *hostp;

array_foreach(mail_hosts_get(dir->mail_hosts), hostp)

director_send_host(dir, dir->self_host, NULL, *hostp);

}

void director_update_host(struct director *dir, struct director_host *src,

struct director_host *orig_src,

struct mail_host *host)

{

/* update state in case this is the first mail host being added */

director_set_state_changed(dir);

dir_debug("Updating host %s vhost_count=%u "

"down=%d last_updown_change=%ld (hosts_hash=%u)",

net_ip2addr(&host->ip), host->vhost_count, host->down ? 1 : 0,

(long)host->last_updown_change,

mail_hosts_hash(dir->mail_hosts));

director_send_host(dir, src, orig_src, host);

/* mark the host desynced until ring is synced again. except if we're

if (dir->right != NULL || dir->left != NULL)

host->desynced = TRUE;

director_sync(dir);

}

void director_remove_host(struct director *dir, struct director_host *src,

struct director_host *orig_src,

struct mail_host *host)

{

if (src != NULL) {

if (orig_src == NULL) {

orig_src = dir->self_host;

orig_src->last_seq++;

}

director_update_send(dir, src, t_strdup_printf(

"HOST-REMOVE\t%s\t%u\t%u\t%s\n",

net_ip2addr(&orig_src->ip), orig_src->port,

orig_src->last_seq, net_ip2addr(&host->ip)));

}

user_directory_remove_host(dir->users, host);

mail_host_remove(host);

director_sync(dir);

}

void director_flush_host(struct director *dir, struct director_host *src,

struct director_host *orig_src,

struct mail_host *host)

{

if (orig_src == NULL) {

orig_src = dir->self_host;

orig_src->last_seq++;

}

director_update_send(dir, src, t_strdup_printf(

"HOST-FLUSH\t%s\t%u\t%u\t%s\n",

net_ip2addr(&orig_src->ip), orig_src->port, orig_src->last_seq,

net_ip2addr(&host->ip)));

user_directory_remove_host(dir->users, host);

director_sync(dir);

}

void director_update_user(struct director *dir, struct director_host *src,

struct user *user)

{

i_assert(src != NULL);

i_assert(!user->weak);

director_update_send(dir, src, t_strdup_printf("USER\t%u\t%s\n",

user->username_hash, net_ip2addr(&user->host->ip)));

}

void director_update_user_weak(struct director *dir, struct director_host *src,

struct director_connection *src_conn,

struct director_host *orig_src,

struct user *user)

{

const char *cmd;

i_assert(src != NULL);

i_assert(user->weak);

if (orig_src == NULL) {

orig_src = dir->self_host;

orig_src->last_seq++;

}

cmd = t_strdup_printf("USER-WEAK\t%s\t%u\t%u\t%u\t%s\n",

net_ip2addr(&orig_src->ip), orig_src->port, orig_src->last_seq,

user->username_hash, net_ip2addr(&user->host->ip));

if (src != dir->self_host && dir->left != NULL && dir->right != NULL &&

director_connection_get_host(dir->left) ==

director_connection_get_host(dir->right)) {

/* only two directors in this ring and we're forwarding

if (dir->right == src_conn)

director_connection_send(dir->left, cmd);

else if (dir->left == src_conn)

director_connection_send(dir->right, cmd);

else

i_unreached();

} else {

director_update_send(dir, src, cmd);

}

}

static void

director_flush_user_continue(int result, struct director_kill_context *ctx)

{

struct director *dir = ctx->dir;

struct user *user =

user_directory_lookup(dir->users, ctx->username_hash);

ctx->callback_pending = FALSE;

if (result == 0) {

struct istream *is = iostream_temp_finish(&ctx->reply, (size_t)-1);

char *data;

i_stream_set_return_partial_line(is, TRUE);

data = i_stream_read_next_line(is);

i_error("%s: Failed to flush user hash %u in host %s: %s",

ctx->socket_path,

ctx->username_hash,

net_ip2addr(&ctx->host_ip),

data == NULL ? "(no output to stdout)" : data);

while((data = i_stream_read_next_line(is)) != NULL) {

i_error("%s: Failed to flush user hash %u in host %s: %s",

ctx->socket_path,

ctx->username_hash,

net_ip2addr(&ctx->host_ip),

data);

}

i_stream_unref(&is);

} else {

o_stream_unref(&ctx->reply);

}

program_client_destroy(&ctx->pclient);

if (!DIRECTOR_KILL_CONTEXT_IS_VALID(user, ctx)) {

/* user was already freed - ignore */

dir_debug("User %u freed while flushing, result=%d",

ctx->username_hash, result);

i_assert(ctx->to_move == NULL);

i_free(ctx);

} else {

/* ctx is freed later via user->kill_ctx */

dir_debug("Flushing user %u finished, result=%d",

ctx->username_hash, result);

director_user_kill_finish_delayed(dir, user, result == 1);

}

}

static void

director_flush_user(struct director *dir, struct user *user)

{

struct director_kill_context *ctx = user->kill_ctx;

struct var_expand_table tab[] = {

{ 'i', net_ip2addr(&user->host->ip), "ip" },

{ 'h', user->host->hostname, "host" },

{ '\0', NULL, NULL }

};

/* Execute flush script, if set. Only the director that started the

if (*dir->set->director_flush_socket == '\0' ||

!ctx->kill_is_self_initiated) {

director_user_kill_finish_delayed(dir, user, FALSE);

return;

}

ctx->host_ip = user->host->ip;

string_t *s_sock = str_new(default_pool, 32);

var_expand(s_sock, dir->set->director_flush_socket, tab);

ctx->socket_path = str_free_without_data(&s_sock);

const char *error;

struct program_client_settings set = {

.client_connect_timeout_msecs = 10000,

.dns_client_socket_path = DIRECTOR_DNS_SOCKET_PATH,

};

restrict_access_init(&set.restrict_set);

const char *const args[] = {"FLUSH",

t_strdup_printf("%u", user->username_hash), NULL};

ctx->kill_state = USER_KILL_STATE_FLUSHING;

dir_debug("Flushing user %u via %s", user->username_hash,

ctx->socket_path);

if ((program_client_create(ctx->socket_path, args, &set, FALSE,

&ctx->pclient, &error)) != 0) {

i_error("%s: Failed to flush user hash %u in host %s: %s",

ctx->socket_path,

user->username_hash,

net_ip2addr(&user->host->ip),

error);

director_flush_user_continue(0, ctx);

return;

}

ctx->reply =

iostream_temp_create_named("/tmp", 0,

t_strdup_printf("flush response from %s",

net_ip2addr(&user->host->ip)));

o_stream_set_no_error_handling(ctx->reply, TRUE);

program_client_set_output(ctx->pclient, ctx->reply);

ctx->callback_pending = TRUE;

program_client_run_async(ctx->pclient, director_flush_user_continue, ctx);

}

static void director_user_move_free(struct user *user)

{

struct director *dir = user->kill_ctx->dir;

struct director_kill_context *kill_ctx = user->kill_ctx;

i_assert(kill_ctx != NULL);

dir_debug("User %u move finished at state=%s", user->username_hash,

user_kill_state_names[kill_ctx->kill_state]);

if (kill_ctx->to_move != NULL)

timeout_remove(&kill_ctx->to_move);

i_free(kill_ctx->socket_path);

i_free(kill_ctx);

user->kill_ctx = NULL;

i_assert(dir->users_moving_count > 0);

dir->users_moving_count--;

dir->state_change_callback(dir);

}

static void

director_user_kill_finish_delayed_to(struct user *user)

{

i_assert(user->kill_ctx != NULL);

i_assert(user->kill_ctx->kill_state == USER_KILL_STATE_DELAY);

director_user_move_free(user);

}

static void

director_user_kill_finish_delayed(struct director *dir, struct user *user,

bool skip_delay)

{

if (skip_delay) {

user->kill_ctx->kill_state = USER_KILL_STATE_NONE;

director_user_move_free(user);

return;

}

user->kill_ctx->kill_state = USER_KILL_STATE_DELAY;

/* wait for a while for the kills to finish in the backend server,

timeout_remove(&user->kill_ctx->to_move);

user->kill_ctx->to_move =

timeout_add(dir->set->director_user_kick_delay * 1000,

director_user_kill_finish_delayed_to, user);

}

static void

director_finish_user_kill(struct director *dir, struct user *user, bool self)

{

struct director_kill_context *kill_ctx = user->kill_ctx;

i_assert(kill_ctx != NULL);

i_assert(kill_ctx->kill_state != USER_KILL_STATE_FLUSHING);

i_assert(kill_ctx->kill_state != USER_KILL_STATE_DELAY);

if (dir->right == NULL) {

/* we're alone */

director_flush_user(dir, user);

} else if (self ||

kill_ctx->kill_state == USER_KILL_STATE_KILLING_NOTIFY_RECEIVED) {

director_connection_send(dir->right, t_strdup_printf(

"USER-KILLED\t%u\n", user->username_hash));

kill_ctx->kill_state = USER_KILL_STATE_KILLED_WAITING_FOR_EVERYONE;

} else {

i_assert(kill_ctx->kill_state == USER_KILL_STATE_KILLING);

kill_ctx->kill_state = USER_KILL_STATE_KILLED_WAITING_FOR_NOTIFY;

}

}

static void director_user_kill_fail_throttled(unsigned int new_events_count,

void *context ATTR_UNUSED)

{

i_error("Failed to kill %u users' connections", new_events_count);

}

static void director_kill_user_callback(enum ipc_client_cmd_state state,

const char *data, void *context)

{

struct director_kill_context *ctx = context;

struct user *user;

/* this is an asynchronous notification about user being killed.

switch (state) {

case IPC_CLIENT_CMD_STATE_REPLY:

/* shouldn't get here. the command reply isn't finished yet. */

return;

case IPC_CLIENT_CMD_STATE_OK:

break;

case IPC_CLIENT_CMD_STATE_ERROR:

if (log_throttle_accept(user_kill_fail_throttle)) {

i_error("Failed to kill user %u connections: %s",

ctx->username_hash, data);

}

/* we can't really do anything but continue anyway */

break;

}

ctx->callback_pending = FALSE;

user = user_directory_lookup(ctx->dir->users, ctx->username_hash);

if (!DIRECTOR_KILL_CONTEXT_IS_VALID(user, ctx)) {

/* user was already freed - ignore */

i_assert(ctx->to_move == NULL);

i_free(ctx);

} else {

i_assert(ctx->kill_state == USER_KILL_STATE_KILLING ||

ctx->kill_state == USER_KILL_STATE_KILLING_NOTIFY_RECEIVED);

/* we were still waiting for the kill notification */

director_finish_user_kill(ctx->dir, user, ctx->kill_is_self_initiated);

}

}

static void director_user_move_throttled(unsigned int new_events_count,

void *context ATTR_UNUSED)

{

i_error("%u users' move timed out, their state may now be inconsistent",

new_events_count);

}

static void director_user_move_timeout(struct user *user)

{

i_assert(user->kill_ctx != NULL);

i_assert(user->kill_ctx->kill_state != USER_KILL_STATE_FLUSHING);

i_assert(user->kill_ctx->kill_state != USER_KILL_STATE_DELAY);

if (log_throttle_accept(user_move_throttle)) {

i_error("Finishing user %u move timed out, "

"its state may now be inconsistent (state=%s)",

user->username_hash,

user_kill_state_names[user->kill_ctx->kill_state]);

}

director_user_move_free(user);

}

static void

director_kill_user(struct director *dir, struct director_host *src,

struct user *user, struct mail_host *old_host)

{

struct director_kill_context *ctx;

const char *cmd;

if (USER_IS_BEING_KILLED(user)) {

/* User is being moved again before the previous move

dir_debug("User %u move restarted - previous kill_state=%s",

user->username_hash,

user_kill_state_names[user->kill_ctx->kill_state]);

return;

}

user->kill_ctx = ctx = i_new(struct director_kill_context, 1);

ctx->dir = dir;

ctx->username_hash = user->username_hash;

ctx->kill_is_self_initiated = src->self;

dir->users_moving_count++;

ctx->to_move = timeout_add(DIRECTOR_USER_MOVE_TIMEOUT_MSECS,

director_user_move_timeout, user);

ctx->kill_state = USER_KILL_STATE_KILLING;

if (old_host != NULL) {

cmd = t_strdup_printf("proxy\t*\tKICK-DIRECTOR-HASH\t%u",

user->username_hash);

ctx->callback_pending = TRUE;

ipc_client_cmd(dir->ipc_proxy, cmd,

director_kill_user_callback, ctx);

} else {

/* we didn't even know about the user before now.

director_finish_user_kill(ctx->dir, user,

ctx->kill_is_self_initiated);

}

}

void director_move_user(struct director *dir, struct director_host *src,

struct director_host *orig_src,

unsigned int username_hash, struct mail_host *host)

{

struct user *user;

/* 1. move this user's host, and set its "killing" flag to delay all of

user = user_directory_lookup(dir->users, username_hash);

if (user == NULL) {

user = user_directory_add(dir->users, username_hash,

host, ioloop_time);

director_kill_user(dir, src, user, NULL);

} else {

struct mail_host *old_host = user->host;

if (user->host == host) {

/* user is already in this host */

return;

}

user->host->user_count--;

user->host = host;

user->host->user_count++;

user->timestamp = ioloop_time;

director_kill_user(dir, src, user, old_host);

}

if (orig_src == NULL) {

orig_src = dir->self_host;

orig_src->last_seq++;

}

director_update_send(dir, src, t_strdup_printf(

"USER-MOVE\t%s\t%u\t%u\t%u\t%s\n",

net_ip2addr(&orig_src->ip), orig_src->port, orig_src->last_seq,

user->username_hash, net_ip2addr(&user->host->ip)));

}

static void

director_kick_user_callback(enum ipc_client_cmd_state state ATTR_UNUSED,

const char *data ATTR_UNUSED,

void *context ATTR_UNUSED)

{

}

void director_kick_user(struct director *dir, struct director_host *src,

struct director_host *orig_src, const char *username)

{

string_t *cmd = t_str_new(64);

str_append(cmd, "proxy\t*\tKICK\t");

str_append_tabescaped(cmd, username);

ipc_client_cmd(dir->ipc_proxy, str_c(cmd),

director_kick_user_callback, (void *)NULL);

if (orig_src == NULL) {

orig_src = dir->self_host;

orig_src->last_seq++;

}

str_printfa(cmd, "USER-KICK\t%s\t%u\t%u\t",

net_ip2addr(&orig_src->ip), orig_src->port, orig_src->last_seq);

str_append_tabescaped(cmd, username);

str_append_c(cmd, '\n');

director_update_send_version(dir, src, DIRECTOR_VERSION_USER_KICK, str_c(cmd));

}

void director_kick_user_alt(struct director *dir, struct director_host *src,

struct director_host *orig_src,

const char *field, const char *value)

{

string_t *cmd = t_str_new(64);

str_append(cmd, "proxy\t*\tKICK-ALT\t");

str_append_tabescaped(cmd, field);

str_append_c(cmd, '\t');

str_append_tabescaped(cmd, value);

ipc_client_cmd(dir->ipc_proxy, str_c(cmd),

director_kick_user_callback, (void *)NULL);

if (orig_src == NULL) {

orig_src = dir->self_host;

orig_src->last_seq++;

}

str_printfa(cmd, "USER-KICK-ALT\t%s\t%u\t%u\t",

net_ip2addr(&orig_src->ip), orig_src->port, orig_src->last_seq);

str_append_tabescaped(cmd, field);

str_append_c(cmd, '\t');

str_append_tabescaped(cmd, value);

str_append_c(cmd, '\n');

director_update_send_version(dir, src, DIRECTOR_VERSION_USER_KICK_ALT, str_c(cmd));

}

void director_kick_user_hash(struct director *dir, struct director_host *src,

struct director_host *orig_src,

unsigned int username_hash,

const struct ip_addr *except_ip)

{

const char *cmd;

cmd = t_strdup_printf("proxy\t*\tKICK-DIRECTOR-HASH\t%u\t%s",

username_hash, net_ip2addr(except_ip));

ipc_client_cmd(dir->ipc_proxy, cmd,

director_kick_user_callback, (void *)NULL);

if (orig_src == NULL) {

orig_src = dir->self_host;

orig_src->last_seq++;

}

cmd = t_strdup_printf("USER-KICK-HASH\t%s\t%u\t%u\t%u\t%s\n",

net_ip2addr(&orig_src->ip), orig_src->port, orig_src->last_seq,

username_hash, net_ip2addr(except_ip));

director_update_send_version(dir, src, DIRECTOR_VERSION_USER_KICK, cmd);

}

static void

director_send_user_killed_everywhere(struct director *dir,

struct director_host *src,

struct director_host *orig_src,

unsigned int username_hash)

{

if (orig_src == NULL) {

orig_src = dir->self_host;

orig_src->last_seq++;

}

director_update_send(dir, src, t_strdup_printf(

"USER-KILLED-EVERYWHERE\t%s\t%u\t%u\t%u\n",

net_ip2addr(&orig_src->ip), orig_src->port, orig_src->last_seq,

username_hash));

}

void director_user_killed(struct director *dir, unsigned int username_hash)

{

struct user *user;

user = user_directory_lookup(dir->users, username_hash);

if (user == NULL || !USER_IS_BEING_KILLED(user))

return;

switch (user->kill_ctx->kill_state) {

case USER_KILL_STATE_KILLING:

user->kill_ctx->kill_state = USER_KILL_STATE_KILLING_NOTIFY_RECEIVED;

break;

case USER_KILL_STATE_KILLED_WAITING_FOR_NOTIFY:

director_finish_user_kill(dir, user, TRUE);

break;

case USER_KILL_STATE_KILLING_NOTIFY_RECEIVED:

dir_debug("User %u kill_state=%s - ignoring USER-KILLED",

username_hash, user_kill_state_names[user->kill_ctx->kill_state]);

break;

case USER_KILL_STATE_NONE:

case USER_KILL_STATE_FLUSHING:

case USER_KILL_STATE_DELAY:

/* move restarted. state=none can also happen if USER-MOVE was

director_send_user_killed_everywhere(dir, dir->self_host, NULL,

username_hash);

break;

case USER_KILL_STATE_KILLED_WAITING_FOR_EVERYONE:

director_user_killed_everywhere(dir, dir->self_host,

NULL, username_hash);

break;

}

}

void director_user_killed_everywhere(struct director *dir,

struct director_host *src,

struct director_host *orig_src,

unsigned int username_hash)

{

struct user *user;

user = user_directory_lookup(dir->users, username_hash);

if (user == NULL) {

dir_debug("User %u no longer exists - ignoring USER-KILLED-EVERYWHERE",

username_hash);

return;

}

if (!USER_IS_BEING_KILLED(user)) {

dir_debug("User %u is no longer being killed - ignoring USER-KILLED-EVERYWHERE",

username_hash);

return;

}

if (user->kill_ctx->kill_state != USER_KILL_STATE_KILLED_WAITING_FOR_EVERYONE) {

dir_debug("User %u kill_state=%s - ignoring USER-KILLED-EVERYWHERE",

username_hash, user_kill_state_names[user->kill_ctx->kill_state]);

return;

}

director_flush_user(dir, user);

director_send_user_killed_everywhere(dir, src, orig_src, username_hash);

}

static void director_state_callback_timeout(struct director *dir)

{

timeout_remove(&dir->to_callback);

dir->state_change_callback(dir);

}

void director_set_state_changed(struct director *dir)

{

/* we may get called to here from various places. use a timeout to

if (dir->to_callback == NULL) {

dir->to_callback =

timeout_add(0, director_state_callback_timeout, dir);

}

}

void director_update_send(struct director *dir, struct director_host *src,

const char *cmd)

{

director_update_send_version(dir, src, 0, cmd);

}

void director_update_send_version(struct director *dir,

struct director_host *src,

unsigned int min_version, const char *cmd)

{

struct director_connection *const *connp;

i_assert(src != NULL);

array_foreach(&dir->connections, connp) {

if (director_connection_get_host(*connp) != src &&

director_connection_get_minor_version(*connp) >= min_version)

director_connection_send(*connp, cmd);

}

}

static void director_user_freed(struct user *user)

{

if (user->kill_ctx != NULL) {

/* director_user_expire is very short. user expired before

if (user->kill_ctx->callback_pending) {

/* kill_ctx is used as a callback parameter.

if (user->kill_ctx->to_move != NULL)

timeout_remove(&user->kill_ctx->to_move);

} else {

director_user_move_free(user);

}

}

}

struct director *

director_init(const struct director_settings *set,

const struct ip_addr *listen_ip, in_port_t listen_port,

director_state_change_callback_t *callback)

{

struct director *dir;

dir = i_new(struct director, 1);

dir->set = set;

dir->self_port = listen_port;

dir->self_ip = *listen_ip;

dir->state_change_callback = callback;