--- a/.hgignore	Thu Dec 29 11:19:52 2011 +0200
+++ b/.hgignore	Thu Dec 29 14:43:45 2011 +0200
@@ -66,7 +66,6 @@
 src/dns/dns-client
 src/doveadm/doveadm
 src/doveadm/doveadm-server
-src/dsync/dsync
 src/imap-login/imap-login
 src/imap/imap
 src/indexer/indexer

--- a/configure.in	Thu Dec 29 11:19:52 2011 +0200
+++ b/configure.in	Thu Dec 29 14:43:45 2011 +0200
@@ -2750,7 +2750,7 @@
 src/auth/Makefile
 src/config/Makefile
 src/doveadm/Makefile
-src/dsync/Makefile
+src/doveadm/dsync/Makefile
 src/lda/Makefile
 src/log/Makefile
 src/lmtp/Makefile

--- a/src/Makefile.am	Thu Dec 29 11:19:52 2011 +0200
+++ b/src/Makefile.am	Thu Dec 29 14:43:45 2011 +0200
@@ -41,7 +41,6 @@
 	director \
 	util \
 	doveadm \
-	dsync \
 	ssl-params \
 	stats \
 	plugins

--- a/src/doveadm/Makefile.am	Thu Dec 29 11:19:52 2011 +0200
+++ b/src/doveadm/Makefile.am	Thu Dec 29 14:43:45 2011 +0200
@@ -1,6 +1,8 @@
 doveadm_moduledir = $(moduledir)/doveadm
 pkglibexecdir = $(libexecdir)/dovecot
 
+SUBDIRS = dsync
+
 bin_PROGRAMS = doveadm
 pkglibexec_PROGRAMS = doveadm-server
 
@@ -35,6 +37,7 @@
 	../lib-otp/libotp.a
 
 libs = \
+	dsync/libdsync.a \
 	$(LIBDOVECOT_STORAGE) \
 	$(unused_objects)
 
@@ -122,3 +125,7 @@
 	doveadm-settings.h \
 	doveadm-util.h \
 	doveadm-who.h
+
+install-exec-local:
+	rm -f $(DESTDIR)$(bindir)/dsync
+	$(LN_S) doveadm $(DESTDIR)$(bindir)/dsync

--- a/src/doveadm/client-connection.c	Thu Dec 29 11:19:52 2011 +0200
+++ b/src/doveadm/client-connection.c	Thu Dec 29 14:43:45 2011 +0200
@@ -41,9 +41,6 @@
 			const struct mail_storage_service_input *input,
 			int argc, char *argv[])
 {
-	enum mail_storage_service_flags service_flags =
-		MAIL_STORAGE_SERVICE_FLAG_NO_LOG_INIT |
-		MAIL_STORAGE_SERVICE_FLAG_USERDB_LOOKUP;
 	struct doveadm_mail_cmd_context *ctx;
 	const struct doveadm_mail_cmd *cmd;
 	const char *getopt_args;
@@ -56,12 +53,15 @@
 		return FALSE;
 	}
 
-	if (doveadm_debug)
-		service_flags |= MAIL_STORAGE_SERVICE_FLAG_DEBUG;
-
 	ctx = doveadm_mail_cmd_init(cmd, set);
 	ctx->full_args = (const void *)(argv + 1);
 
+	ctx->service_flags |=
+		MAIL_STORAGE_SERVICE_FLAG_NO_LOG_INIT |
+		MAIL_STORAGE_SERVICE_FLAG_USERDB_LOOKUP;
+	if (doveadm_debug)
+		ctx->service_flags |= MAIL_STORAGE_SERVICE_FLAG_DEBUG;
+
 	getopt_args = t_strconcat("AS:u:", ctx->getopt_args, NULL);
 	while ((c = getopt(argc, argv, getopt_args)) > 0) {
 		switch (c) {
@@ -106,7 +106,10 @@
 	}
 
 	ctx->args = (const void *)argv;
-	doveadm_mail_single_user(ctx, input, service_flags);
+	if (ctx->v.preinit != NULL)
+		ctx->v.preinit(ctx);
+
+	doveadm_mail_single_user(ctx, input);
 	doveadm_mail_server_flush();
 	ctx->v.deinit(ctx);
 	doveadm_print_flush();

--- a/src/doveadm/doveadm-mail.c	Thu Dec 29 11:19:52 2011 +0200
+++ b/src/doveadm/doveadm-mail.c	Thu Dec 29 14:43:45 2011 +0200
@@ -19,6 +19,7 @@
 #include "doveadm.h"
 #include "doveadm-settings.h"
 #include "doveadm-print.h"
+#include "dsync/doveadm-dsync.h"
 #include "doveadm-mail.h"
 
 #include <stdio.h>
@@ -186,7 +187,6 @@
 		       const struct mail_storage_service_input *input,
 		       const char **error_r)
 {
-	struct mail_storage_service_user *service_user;
 	const char *error;
 	int ret;
 
@@ -199,7 +199,7 @@
 		return ret;
 
 	ret = mail_storage_service_lookup(ctx->storage_service, input,
-					  &service_user, &error);
+					  &ctx->cur_service_user, &error);
 	if (ret <= 0) {
 		if (ret < 0) {
 			*error_r = t_strdup_printf("User lookup failed: %s",
@@ -208,31 +208,32 @@
 		return ret;
 	}
 
-	ret = mail_storage_service_next(ctx->storage_service, service_user,
+	ret = mail_storage_service_next(ctx->storage_service,
+					ctx->cur_service_user,
 					&ctx->cur_mail_user);
 	if (ret < 0) {
 		*error_r = "User init failed";
-		mail_storage_service_user_free(&service_user);
+		mail_storage_service_user_free(&ctx->cur_service_user);
 		return ret;
 	}
 
 	ctx->v.run(ctx, ctx->cur_mail_user);
 	mail_user_unref(&ctx->cur_mail_user);
-	mail_storage_service_user_free(&service_user);
+	mail_storage_service_user_free(&ctx->cur_service_user);
 	return 1;
 }
 
 void doveadm_mail_single_user(struct doveadm_mail_cmd_context *ctx,
-			      const struct mail_storage_service_input *input,
-			      enum mail_storage_service_flags service_flags)
+			      const struct mail_storage_service_input *input)
 {
 	const char *error;
 	int ret;
 
 	i_assert(input->username != NULL);
 
+	ctx->cur_username = input->username;
 	ctx->storage_service = mail_storage_service_init(master_service, NULL,
-							 service_flags);
+							 ctx->service_flags);
 	ctx->v.init(ctx, ctx->args);
 	if (hook_doveadm_mail_init != NULL)
 		hook_doveadm_mail_init(ctx);
@@ -251,21 +252,20 @@
 
 static void
 doveadm_mail_all_users(struct doveadm_mail_cmd_context *ctx, char *argv[],
-		       const char *wildcard_user,
-		       enum mail_storage_service_flags service_flags)
+		       const char *wildcard_user)
 {
 	struct mail_storage_service_input input;
 	unsigned int user_idx, user_count, interval, n;
 	const char *user, *error;
 	int ret;
 
-	service_flags |= MAIL_STORAGE_SERVICE_FLAG_USERDB_LOOKUP;
+	ctx->service_flags |= MAIL_STORAGE_SERVICE_FLAG_USERDB_LOOKUP;
 
 	memset(&input, 0, sizeof(input));
 	input.service = "doveadm";
 
 	ctx->storage_service = mail_storage_service_init(master_service, NULL,
-							 service_flags);
+							 ctx->service_flags);
         lib_signals_set_handler(SIGINT, 0, sig_die, NULL);
 	lib_signals_set_handler(SIGTERM, 0, sig_die, NULL);
 
@@ -355,20 +355,19 @@
 static void
 doveadm_mail_cmd(const struct doveadm_mail_cmd *cmd, int argc, char *argv[])
 {
-	enum mail_storage_service_flags service_flags =
-		MAIL_STORAGE_SERVICE_FLAG_NO_LOG_INIT;
 	struct doveadm_mail_cmd_context *ctx;
-	const char *getopt_args, *username, *wildcard_user;
+	const char *getopt_args, *wildcard_user;
 	int c;
 
-	if (doveadm_debug)
-		service_flags |= MAIL_STORAGE_SERVICE_FLAG_DEBUG;
-
 	ctx = doveadm_mail_cmd_init(cmd, doveadm_settings);
 	ctx->full_args = (const void *)(argv + 1);
 
+	ctx->service_flags |= MAIL_STORAGE_SERVICE_FLAG_NO_LOG_INIT;
+	if (doveadm_debug)
+		ctx->service_flags |= MAIL_STORAGE_SERVICE_FLAG_DEBUG;
+
 	getopt_args = t_strconcat("AS:u:", ctx->getopt_args, NULL);
-	username = getenv("USER");
+	ctx->cur_username = getenv("USER");
 	wildcard_user = NULL;
 	while ((c = getopt(argc, argv, getopt_args)) > 0) {
 		switch (c) {
@@ -381,12 +380,14 @@
 				doveadm_settings->doveadm_worker_count = 1;
 			break;
 		case 'u':
-			service_flags |=
+			ctx->service_flags |=
 				MAIL_STORAGE_SERVICE_FLAG_USERDB_LOOKUP;
-			username = optarg;
-			if (strchr(username, '*') != NULL ||
-			    strchr(username, '?') != NULL)
-				wildcard_user = username;
+			ctx->cur_username = optarg;
+			if (strchr(ctx->cur_username, '*') != NULL ||
+			    strchr(ctx->cur_username, '?') != NULL) {
+				wildcard_user = ctx->cur_username;
+				ctx->cur_username = NULL;
+			}
 			break;
 		default:
 			if (ctx->v.parse_arg == NULL ||
@@ -400,6 +401,8 @@
 			cmd->name, argv[0]);
 	}
 	ctx->args = (const void *)argv;
+	if (ctx->v.preinit != NULL)
+		ctx->v.preinit(ctx);
 
 	ctx->iterate_single_user =
 		!ctx->iterate_all_users && wildcard_user == NULL;
@@ -412,16 +415,16 @@
 	if (ctx->iterate_single_user) {
 		struct mail_storage_service_input input;
 
-		if (username == NULL)
+		if (ctx->cur_username == NULL)
 			i_fatal("USER environment is missing and -u option not used");
 
 		memset(&input, 0, sizeof(input));
 		input.service = "doveadm";
-		input.username = username;
-		doveadm_mail_single_user(ctx, &input, service_flags);
+		input.username = ctx->cur_username;
+		doveadm_mail_single_user(ctx, &input);
 	} else {
-		service_flags |= MAIL_STORAGE_SERVICE_FLAG_TEMP_PRIV_DROP;
-		doveadm_mail_all_users(ctx, argv, wildcard_user, service_flags);
+		ctx->service_flags |= MAIL_STORAGE_SERVICE_FLAG_TEMP_PRIV_DROP;
+		doveadm_mail_all_users(ctx, argv, wildcard_user);
 	}
 	if (ctx->search_args != NULL)
 		mail_search_args_unref(&ctx->search_args);
@@ -432,6 +435,8 @@
 	/* service deinit unloads mail plugins, so do it late */
 	mail_storage_service_deinit(&ctx->storage_service);
 
+	if (ctx->exit_code != 0)
+		exit(ctx->exit_code);
 	if (ctx->failed)
 		exit(FATAL_DEFAULT);
 	pool_unref(&ctx->pool);
@@ -579,7 +584,10 @@
 	&cmd_mailbox_rename,
 	&cmd_mailbox_subscribe,
 	&cmd_mailbox_unsubscribe,
-	&cmd_mailbox_status
+	&cmd_mailbox_status,
+	&cmd_dsync_backup,
+	&cmd_dsync_mirror,
+	&cmd_dsync_server
 };
 
 void doveadm_mail_init(void)

--- a/src/doveadm/doveadm-mail.h	Thu Dec 29 11:19:52 2011 +0200
+++ b/src/doveadm/doveadm-mail.h	Thu Dec 29 14:43:45 2011 +0200
@@ -4,17 +4,15 @@
 #include <stdio.h>
 #include "doveadm-util.h"
 #include "module-context.h"
+#include "mail-storage-service.h"
 
-enum mail_storage_service_flags;
 struct mailbox;
 struct mail_user;
-struct mail_storage_service_ctx;
-struct mail_storage_service_input;
-struct mail_storage_service_user;
 struct doveadm_mail_cmd_context;
 
 struct doveadm_mail_cmd_vfuncs {
 	bool (*parse_arg)(struct doveadm_mail_cmd_context *ctx,int c);
+	void (*preinit)(struct doveadm_mail_cmd_context *ctx);
 	void (*init)(struct doveadm_mail_cmd_context *ctx,
 		     const char *const args[]);
 	int (*get_next_user)(struct doveadm_mail_cmd_context *ctx,
@@ -42,15 +40,21 @@
 
 	const char *getopt_args;
 	const struct doveadm_settings *set;
+	enum mail_storage_service_flags service_flags;
 	struct mail_storage_service_ctx *storage_service;
 	/* search args aren't set for all mail commands */
 	struct mail_search_args *search_args;
 
+	const char *cur_username;
+	struct mail_storage_service_user *cur_service_user;
 	struct mail_user *cur_mail_user;
 	struct doveadm_mail_cmd_vfuncs v;
 
 	ARRAY_DEFINE(module_contexts, union doveadm_mail_cmd_module_context *);
 
+	/* if non-zero, exit with this code */
+	int exit_code;
+
 	/* We're handling only a single user */
 	unsigned int iterate_single_user:1;
 	/* We're going through all users (not set for wildcard usernames) */
@@ -86,8 +90,7 @@
 doveadm_mail_cmd_init(const struct doveadm_mail_cmd *cmd,
 		      const struct doveadm_settings *set);
 void doveadm_mail_single_user(struct doveadm_mail_cmd_context *ctx,
-			      const struct mail_storage_service_input *input,
-			      enum mail_storage_service_flags service_flags);
+			      const struct mail_storage_service_input *input);
 int doveadm_mail_server_user(struct doveadm_mail_cmd_context *ctx,
 			     const struct mail_storage_service_input *input,
 			     const char **error_r);

--- a/src/doveadm/doveadm-settings.c	Thu Dec 29 11:19:52 2011 +0200
+++ b/src/doveadm/doveadm-settings.c	Thu Dec 29 14:43:45 2011 +0200
@@ -60,6 +60,7 @@
 	DEF(SET_UINT, doveadm_proxy_port),
 	DEF(SET_STR, doveadm_password),
 	DEF(SET_STR, doveadm_allowed_commands),
+	DEF(SET_STR, dsync_alt_char),
 
 	{ SET_STRLIST, "plugin", offsetof(struct doveadm_settings, plugin_envs), NULL },
 
@@ -75,6 +76,7 @@
 	.doveadm_proxy_port = 0,
 	.doveadm_password = "",
 	.doveadm_allowed_commands = "",
+	.dsync_alt_char = "_",
 
 	.plugin_envs = ARRAY_INIT
 };

--- a/src/doveadm/doveadm-settings.h	Thu Dec 29 11:19:52 2011 +0200
+++ b/src/doveadm/doveadm-settings.h	Thu Dec 29 14:43:45 2011 +0200
@@ -10,6 +10,7 @@
 	unsigned int doveadm_proxy_port;
 	const char *doveadm_password;
 	const char *doveadm_allowed_commands;
+	const char *dsync_alt_char;
 
 	ARRAY_DEFINE(plugin_envs, const char *);
 };

--- a/src/doveadm/doveadm.c	Thu Dec 29 11:19:52 2011 +0200
+++ b/src/doveadm/doveadm.c	Thu Dec 29 14:43:45 2011 +0200
@@ -12,6 +12,7 @@
 #include "doveadm-dump.h"
 #include "doveadm-mail.h"
 #include "doveadm-settings.h"
+#include "dsync/doveadm-dsync.h"
 #include "doveadm.h"
 
 #include <stdlib.h>
@@ -274,6 +275,8 @@
 	bool quick_init = FALSE;
 	int c;
 
+	doveadm_dsync_main(&argc, &argv);
+
 	/* "+" is GNU extension to stop at the first non-option.
 	   others just accept -+ option. */
 	master_service = master_service_init("doveadm", service_flags,

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/doveadm/dsync/Makefile.am	Thu Dec 29 14:43:45 2011 +0200
@@ -0,0 +1,80 @@
+noinst_LIBRARIES = libdsync.a
+
+AM_CPPFLAGS = \
+	-I$(top_srcdir)/src/lib \
+	-I$(top_srcdir)/src/lib-test \
+	-I$(top_srcdir)/src/lib-settings \
+	-I$(top_srcdir)/src/lib-master \
+	-I$(top_srcdir)/src/lib-mail \
+	-I$(top_srcdir)/src/lib-imap \
+	-I$(top_srcdir)/src/lib-index \
+	-I$(top_srcdir)/src/lib-storage \
+	-I$(top_srcdir)/src/doveadm
+
+libs = \
+	$(LIBDOVECOT_STORAGE)
+
+libdsync_a_SOURCES = \
+	doveadm-dsync.c \
+	dsync-brain.c \
+	dsync-brain-msgs.c \
+	dsync-brain-msgs-new.c \
+	dsync-data.c \
+	dsync-proxy.c \
+	dsync-proxy-client.c \
+	dsync-proxy-server.c \
+	dsync-proxy-server-cmd.c \
+	dsync-worker.c \
+	dsync-worker-local.c
+
+noinst_HEADERS = \
+	dsync-brain.h \
+	dsync-brain-private.h \
+	dsync-data.h \
+	dsync-proxy.h \
+	dsync-proxy-server.h \
+	dsync-worker.h \
+	dsync-worker-private.h \
+	test-dsync-common.h \
+	test-dsync-worker.h
+
+test_programs = \
+	test-dsync-brain \
+	test-dsync-brain-msgs \
+	test-dsync-proxy \
+	test-dsync-proxy-server-cmd
+
+noinst_PROGRAMS = $(test_programs)
+
+test_libs = \
+	../../lib-test/libtest.la \
+	../../lib-mail/libmail.la \
+	../../lib-imap/libimap.la \
+	../../lib-charset/libcharset.la \
+	../../lib/liblib.la
+
+test_ldadd = \
+	$(test_libs) \
+	$(LIBICONV)
+
+test_dsync_brain_SOURCES = test-dsync-brain.c test-dsync-worker.c test-dsync-common.c
+test_dsync_brain_LDADD = dsync-data.o dsync-brain.o dsync-worker.o $(test_ldadd)
+test_dsync_brain_DEPENDENCIES = dsync-data.o dsync-brain.o dsync-worker.o $(test_libs)
+
+test_dsync_brain_msgs_SOURCES = test-dsync-brain-msgs.c test-dsync-worker.c test-dsync-common.c
+test_dsync_brain_msgs_LDADD = dsync-data.o dsync-brain-msgs.o dsync-worker.o $(test_ldadd)
+test_dsync_brain_msgs_DEPENDENCIES = dsync-data.o dsync-brain-msgs.o dsync-worker.o $(test_libs)
+
+test_dsync_proxy_SOURCES = test-dsync-proxy.c test-dsync-common.c
+test_dsync_proxy_LDADD = dsync-proxy.o dsync-data.o $(test_ldadd)
+test_dsync_proxy_DEPENDENCIES = dsync-proxy.o dsync-data.o $(test_libs)
+
+test_dsync_proxy_server_cmd_SOURCES = test-dsync-proxy-server-cmd.c test-dsync-worker.c test-dsync-common.c
+test_dsync_proxy_server_cmd_LDADD = dsync-worker.o dsync-proxy.o dsync-proxy-server-cmd.o dsync-data.o $(test_ldadd)
+test_dsync_proxy_server_cmd_DEPENDENCIES = dsync-worker.o dsync-proxy.o dsync-proxy-server-cmd.o dsync-data.o $(test_libs)
+
+check: check-am check-test
+check-test: all-am
+	for bin in $(test_programs); do \
+	  if ! $(RUN_TEST) ./$$bin; then exit 1; fi; \
+	done

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/doveadm/dsync/doveadm-dsync.c	Thu Dec 29 14:43:45 2011 +0200
@@ -0,0 +1,494 @@
+/* Copyright (c) 2009-2011 Dovecot authors, see the included COPYING file */
+
+#include "lib.h"
+#include "lib-signals.h"
+#include "array.h"
+#include "execv-const.h"
+#include "settings-parser.h"
+#include "master-service.h"
+#include "mail-storage-service.h"
+#include "mail-user.h"
+#include "mail-namespace.h"
+#include "doveadm-settings.h"
+#include "doveadm-mail.h"
+#include "dsync-brain.h"
+#include "dsync-worker.h"
+#include "dsync-proxy-server.h"
+#include "doveadm-dsync.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <ctype.h>
+
+struct dsync_cmd_context {
+	struct doveadm_mail_cmd_context ctx;
+	enum dsync_brain_flags brain_flags;
+	const char *mailbox;
+
+	const char *const *remote_cmd_args;
+	const char *local_location;
+
+	int fd_in, fd_out;
+
+	unsigned int reverse_workers:1;
+};
+
+static const char *ssh_cmd = "ssh";
+
+static void run_cmd(const char *const *args, int *fd_in_r, int *fd_out_r)
+{
+	int fd_in[2], fd_out[2];
+
+	if (pipe(fd_in) < 0 || pipe(fd_out) < 0)
+		i_fatal("pipe() failed: %m");
+
+	switch (fork()) {
+	case -1:
+		i_fatal("fork() failed: %m");
+		break;
+	case 0:
+		/* child, which will execute the proxy server. stdin/stdout
+		   goes to pipes which we'll pass to proxy client. */
+		if (dup2(fd_in[0], STDIN_FILENO) < 0 ||
+		    dup2(fd_out[1], STDOUT_FILENO) < 0)
+			i_fatal("dup2() failed: %m");
+
+		(void)close(fd_in[0]);
+		(void)close(fd_in[1]);
+		(void)close(fd_out[0]);
+		(void)close(fd_out[1]);
+
+		execvp_const(args[0], args);
+		break;
+	default:
+		/* parent */
+		(void)close(fd_in[0]);
+		(void)close(fd_out[1]);
+		*fd_in_r = fd_out[0];
+		*fd_out_r = fd_in[1];
+		break;
+	}
+}
+
+static void
+mirror_get_remote_cmd_line(const char *const *argv,
+			   const char *const **cmd_args_r)
+{
+	ARRAY_TYPE(const_string) cmd_args;
+	unsigned int i;
+	const char *p;
+
+	t_array_init(&cmd_args, 16);
+	for (i = 0; argv[i] != NULL; i++) {
+		p = argv[i];
+		array_append(&cmd_args, &p, 1);
+	}
+
+	p = strchr(argv[0], '/');
+	if (p == NULL) p = argv[0];
+	if (strstr(p, "dsync") == NULL) {
+		/* we're executing doveadm (not dsync) */
+		p = "dsync"; array_append(&cmd_args, &p, 1);
+	}
+	p = "server"; array_append(&cmd_args, &p, 1);
+	(void)array_append_space(&cmd_args);
+	*cmd_args_r = array_idx(&cmd_args, 0);
+}
+
+static bool mirror_get_remote_cmd(const char *const *argv, const char *user,
+				  const char *const **cmd_args_r)
+{
+	ARRAY_TYPE(const_string) cmd_args;
+	const char *p, *host;
+
+	if (argv[1] != NULL) {
+		/* more than one parameter, so it contains a full command
+		   (e.g. ssh host dsync) */
+		mirror_get_remote_cmd_line(argv, cmd_args_r);
+		return TRUE;
+	}
+
+	/* if it begins with /[a-z0-9]+:/, it's a mail location
+	   (e.g. mdbox:~/mail) */
+	for (p = argv[0]; *p != '\0'; p++) {
+		if (!i_isalnum(*p)) {
+			if (*p == ':')
+				return FALSE;
+			break;
+		}
+	}
+
+	if (strchr(argv[0], ' ') != NULL || strchr(argv[0], '/') != NULL) {
+		/* a) the whole command is in one string. this is mainly for
+		      backwards compatibility.
+		   b) script/path */
+		mirror_get_remote_cmd_line(t_strsplit(argv[0], " "),
+					   cmd_args_r);
+		return TRUE;
+	}
+
+	/* [user@]host */
+	host = strchr(argv[0], '@');
+	if (host != NULL)
+		user = t_strdup_until(argv[0], host++);
+	else
+		host = argv[0];
+
+	/* we'll assume virtual users, so in user@host it really means not to
+	   give ssh a username, but to give dsync -u user parameter. */
+	t_array_init(&cmd_args, 8);
+	array_append(&cmd_args, &ssh_cmd, 1);
+	array_append(&cmd_args, &host, 1);
+	p = "doveadm"; array_append(&cmd_args, &p, 1);
+	p = "dsync"; array_append(&cmd_args, &p, 1);
+	p = "server"; array_append(&cmd_args, &p, 1);
+	if (*user != '\0') {
+		p = "-u"; array_append(&cmd_args, &p, 1);
+		array_append(&cmd_args, &user, 1);
+	}
+	(void)array_append_space(&cmd_args);
+	*cmd_args_r = array_idx(&cmd_args, 0);
+	return TRUE;
+}
+
+static struct dsync_worker *
+cmd_dsync_run_local(struct dsync_cmd_context *ctx, struct mail_user *user)
+{
+	struct mail_user *user2;
+	struct dsync_worker *worker2;
+	struct setting_parser_context *set_parser;
+	const char *set_line, *path1, *path2;
+
+	i_assert(ctx->local_location != NULL);
+
+	ctx->brain_flags |= DSYNC_BRAIN_FLAG_LOCAL;
+	i_set_failure_prefix(t_strdup_printf("dsync(%s): ", user->username));
+
+	/* update mail_location and create another user for the
+	   second location. */
+	set_parser = mail_storage_service_user_get_settings_parser(ctx->ctx.cur_service_user);
+	set_line = t_strconcat("mail_location=", ctx->local_location, NULL);
+	if (settings_parse_line(set_parser, set_line) < 0)
+		i_unreached();
+	if (mail_storage_service_next(ctx->ctx.storage_service,
+				      ctx->ctx.cur_service_user, &user2) < 0)
+		i_fatal("User init failed");
+	user2->admin = TRUE;
+
+	if (mail_namespaces_get_root_sep(user->namespaces) !=
+	    mail_namespaces_get_root_sep(user2->namespaces)) {
+		i_fatal("Mail locations must use the same "
+			"virtual mailbox hierarchy separator "
+			"(specify separator for the default namespace)");
+	}
+	path1 = mailbox_list_get_path(user->namespaces->list, NULL,
+				      MAILBOX_LIST_PATH_TYPE_MAILBOX);
+	path2 = mailbox_list_get_path(user2->namespaces->list, NULL,
+				      MAILBOX_LIST_PATH_TYPE_MAILBOX);
+	if (path1 != NULL && path2 != NULL &&
+	    strcmp(path1, path2) == 0) {
+		i_fatal("Both source and destination mail_location "
+			"points to same directory: %s", path1);
+	}
+
+	worker2 = dsync_worker_init_local(user2, *ctx->ctx.set->dsync_alt_char);
+	mail_user_unref(&user2);
+	return worker2;
+}
+
+static struct dsync_worker *
+cmd_dsync_run_remote(struct dsync_cmd_context *ctx, struct mail_user *user)
+{
+	i_set_failure_prefix(t_strdup_printf("dsync-local(%s): ",
+					     user->username));
+	return dsync_worker_init_proxy_client(ctx->fd_in, ctx->fd_out);
+}
+
+static void
+cmd_dsync_run(struct doveadm_mail_cmd_context *_ctx, struct mail_user *user)
+{
+	struct dsync_cmd_context *ctx = (struct dsync_cmd_context *)_ctx;
+	struct dsync_worker *worker1, *worker2, *workertmp;
+	struct dsync_brain *brain;
+
+	user->admin = TRUE;
+
+	/* create workers */
+	worker1 = dsync_worker_init_local(user, *_ctx->set->dsync_alt_char);
+	if (ctx->remote_cmd_args == NULL)
+		worker2 = cmd_dsync_run_local(ctx, user);
+	else
+		worker2 = cmd_dsync_run_remote(ctx, user);
+	if (ctx->reverse_workers) {
+		workertmp = worker1;
+		worker1 = worker2;
+		worker2 = workertmp;
+	}
+
+	/* create and run the brain */
+	brain = dsync_brain_init(worker1, worker2, ctx->mailbox,
+				 ctx->brain_flags);
+	if (ctx->remote_cmd_args == NULL)
+		dsync_brain_sync_all(brain);
+	else {
+		dsync_brain_sync(brain);
+		if (!dsync_brain_has_failed(brain))
+			io_loop_run(current_ioloop);
+	}
+	/* deinit */
+	if (dsync_brain_has_unexpected_changes(brain)) {
+		i_warning("Mailbox changes caused a desync. "
+			  "You may want to run dsync again.");
+		_ctx->exit_code = 2;
+	}
+	if (dsync_brain_deinit(&brain) < 0)
+		_ctx->exit_code = 1;
+
+	dsync_worker_deinit(&worker1);
+	dsync_worker_deinit(&worker2);
+}
+
+static void cmd_dsync_init(struct doveadm_mail_cmd_context *_ctx,
+			   const char *const args[])
+{
+	struct dsync_cmd_context *ctx = (struct dsync_cmd_context *)_ctx;
+	const char *username = "";
+
+	if (args[0] == NULL)
+		doveadm_mail_help_name("dsync");
+
+	lib_signals_ignore(SIGHUP, TRUE);
+
+	if (doveadm_debug || doveadm_verbose)
+		ctx->brain_flags |= DSYNC_BRAIN_FLAG_VERBOSE;
+
+	/* if we're executing remotely, give -u parameter if we also
+	   did a userdb lookup. this works only when we're handling a
+	   single user */
+	if ((_ctx->service_flags & MAIL_STORAGE_SERVICE_FLAG_USERDB_LOOKUP) != 0 &&
+	    _ctx->cur_username != NULL)
+		username = _ctx->cur_username;
+	if (!mirror_get_remote_cmd(args, username, &ctx->remote_cmd_args)) {
+		/* it's a mail_location */
+		if (args[1] != NULL)
+			doveadm_mail_help_name("dsync");
+		ctx->local_location = args[0];
+	}
+
+	if (ctx->remote_cmd_args != NULL) {
+		/* do this before mail_storage_service_next() in case it
+		   drops process privileges */
+		run_cmd(ctx->remote_cmd_args, &ctx->fd_in, &ctx->fd_out);
+	} else {
+		ctx->fd_in = STDIN_FILENO;
+		ctx->fd_out = STDOUT_FILENO;
+	}
+}
+
+static void cmd_dsync_preinit(struct doveadm_mail_cmd_context *ctx)
+{
+	if ((ctx->service_flags & MAIL_STORAGE_SERVICE_FLAG_USERDB_LOOKUP) == 0)
+		ctx->service_flags |= MAIL_STORAGE_SERVICE_FLAG_NO_CHDIR;
+}
+
+static bool
+cmd_mailbox_dsync_parse_arg(struct doveadm_mail_cmd_context *_ctx, int c)
+{
+	struct dsync_cmd_context *ctx = (struct dsync_cmd_context *)_ctx;
+
+	switch (c) {
+	case 'f':
+		ctx->brain_flags |= DSYNC_BRAIN_FLAG_FULL_SYNC;
+		break;
+	case 'm':
+		ctx->mailbox = optarg;
+		break;
+	case 'R':
+		ctx->reverse_workers = TRUE;
+		break;
+	default:
+		return FALSE;
+	}
+	return TRUE;
+}
+
+static struct doveadm_mail_cmd_context *cmd_dsync_alloc(void)
+{
+	struct dsync_cmd_context *ctx;
+
+	ctx = doveadm_mail_cmd_alloc(struct dsync_cmd_context);
+	ctx->ctx.getopt_args = "fRm:";
+	ctx->ctx.v.parse_arg = cmd_mailbox_dsync_parse_arg;
+	ctx->ctx.v.preinit = cmd_dsync_preinit;
+	ctx->ctx.v.init = cmd_dsync_init;
+	ctx->ctx.v.run = cmd_dsync_run;
+	return &ctx->ctx;
+}
+
+static struct doveadm_mail_cmd_context *cmd_dsync_backup_alloc(void)
+{
+	struct doveadm_mail_cmd_context *_ctx;
+	struct dsync_cmd_context *ctx;
+
+	_ctx = cmd_dsync_alloc();
+	ctx = (struct dsync_cmd_context *)_ctx;
+	ctx->brain_flags |= DSYNC_BRAIN_FLAG_BACKUP;
+	return _ctx;
+}
+
+static void
+cmd_dsync_server_run(struct doveadm_mail_cmd_context *ctx,
+		     struct mail_user *user)
+{
+	struct dsync_proxy_server *server;
+	struct dsync_worker *worker;
+
+	user->admin = TRUE;
+
+	i_set_failure_prefix(t_strdup_printf("dsync-remote(%s): ",
+					     user->username));
+	worker = dsync_worker_init_local(user, *ctx->set->dsync_alt_char);
+	server = dsync_proxy_server_init(STDIN_FILENO, STDOUT_FILENO, worker);
+
+	io_loop_run(current_ioloop);
+
+	dsync_proxy_server_deinit(&server);
+	dsync_worker_deinit(&worker);
+}
+
+static struct doveadm_mail_cmd_context *cmd_dsync_server_alloc(void)
+{
+	struct doveadm_mail_cmd_context *ctx;
+
+	ctx = doveadm_mail_cmd_alloc(struct doveadm_mail_cmd_context);
+	ctx->v.run = cmd_dsync_server_run;
+	return ctx;
+}
+
+struct doveadm_mail_cmd cmd_dsync_mirror = {
+	cmd_dsync_alloc, "dsync mirror", "[-fR] [-m <mailbox>] <dest>"
+};
+struct doveadm_mail_cmd cmd_dsync_backup = {
+	cmd_dsync_backup_alloc, "dsync backup",
+	"[-fR] [-m <mailbox>] <dest>"
+};
+struct doveadm_mail_cmd cmd_dsync_server = {
+	cmd_dsync_server_alloc, "dsync server", NULL
+};
+
+void doveadm_dsync_main(int *_argc, char **_argv[])
+{
+	int argc = *_argc;
+	const char *getopt_str;
+	char **argv = *_argv;
+	char **new_argv, *mailbox = NULL, *alt_char = NULL;
+	char *p, *dup, new_flags[5];
+	int max_argc, src, dest, i, j;
+	bool flag_f = FALSE, flag_R = FALSE, flag_m, flag_C, has_arg;
+
+	p = strrchr(argv[0], '/');
+	if (p == NULL) p = argv[0];
+	if (strstr(p, "dsync") == NULL)
+		return;
+
+	/* @UNSAFE: this is called when the "doveadm" binary is called as
+	   "dsync" (for backwards compatibility) */
+	max_argc = argc + 5;
+	new_argv = calloc(sizeof(char *), max_argc);
+	new_argv[0] = argv[0];
+	dest = 1;
+	getopt_str = master_service_getopt_string();
+
+	/* add global doveadm flags */
+	for (src = 1; src < argc; src++) {
+		if (argv[src][0] != '-')
+			break;
+
+		flag_m = FALSE; flag_C = FALSE; has_arg = FALSE;
+		dup = strdup(argv[src]);
+		for (i = j = 1; argv[src][i] != '\0'; i++) {
+			switch (argv[src][i]) {
+			case 'C':
+				flag_C = TRUE;
+				break;
+			case 'f':
+				flag_f = TRUE;
+				break;
+			case 'R':
+				flag_R = TRUE;
+				break;
+			case 'm':
+				flag_m = TRUE;
+				break;
+			default:
+				p = strchr(getopt_str, argv[src][i]);
+				if (p != NULL && p[1] == ':')
+					has_arg = TRUE;
+				dup[j++] = argv[src][i];
+				break;
+			}
+		}
+		if (j > 1) {
+			dup[j++] = '\0';
+			new_argv[dest++] = dup;
+			if (has_arg && src+1 < argc)
+				new_argv[dest++] = argv[++src];
+		}
+		if (flag_m) {
+			if (src+1 == argc)
+				i_fatal("-m missing parameter");
+			mailbox = argv[++src];
+		}
+		if (flag_C) {
+			if (src+1 == argc)
+				i_fatal("-C missing parameter");
+			alt_char = argv[++src];
+		}
+	}
+	if (alt_char != NULL) {
+		new_argv[dest++] = "-o";
+		new_argv[dest++] =
+			p_strconcat(pool_datastack_create(),
+				    "dsync_alt_char=", alt_char, NULL);
+	}
+
+	new_argv[dest++] = "dsync";
+	if (src < argc) {
+		/* mirror|backup|server */
+		if (strcmp(argv[src], "dsync") == 0) {
+			/* looks like we executed doveconf, which
+			   re-executed ourself with new parameters.
+			   no need to change them anymore. */
+			return;
+		}
+		new_argv[dest++] = argv[src++];
+	}
+
+	/* dsync flags */
+	new_flags[0] = '-'; i = 1;
+	if (flag_f)
+		new_flags[i++] = 'f';
+	if (flag_R)
+		new_flags[i++] = 'R';
+	if (mailbox != NULL)
+		new_flags[i++] = 'm';
+	i_assert((unsigned int)i < sizeof(new_flags));
+	new_flags[i] = '\0';
+
+	if (i > 1) {
+		new_argv[dest++] = strdup(new_flags);
+		if (mailbox != NULL)
+			new_argv[dest++] = mailbox;
+	}
+
+	/* rest of the parameters */
+	for (; src < argc; src++)
+		new_argv[dest++] = argv[src];
+	i_assert(dest < max_argc);
+	new_argv[dest] = NULL;
+
+	*_argc = dest;
+	*_argv = new_argv;
+	optind = 1;
+}

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/doveadm/dsync/doveadm-dsync.h	Thu Dec 29 14:43:45 2011 +0200
@@ -0,0 +1,10 @@
+#ifndef DOVEADM_DSYNC_H
+#define DOVEADM_DSYNC_H
+
+extern struct doveadm_mail_cmd cmd_dsync_mirror;
+extern struct doveadm_mail_cmd cmd_dsync_backup;
+extern struct doveadm_mail_cmd cmd_dsync_server;
+
+void doveadm_dsync_main(int *_argc, char **_argv[]);
+
+#endif

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/doveadm/dsync/dsync-brain-msgs-new.c	Thu Dec 29 14:43:45 2011 +0200
@@ -0,0 +1,392 @@
+/* Copyright (c) 2009-2011 Dovecot authors, see the included COPYING file */
+
+/*
+   This code contains the step 6 explained in dsync-brain-msgs.c:
+   It saves/copies new messages and gives new UIDs for conflicting messages.
+
+   The input is both workers' msg iterators' new_msgs and uid_conflicts
+   variables. They're first sorted by mailbox and secondarily by wanted
+   destination UID. Destination UIDs of conflicts should always be higher
+   than new messages'.
+
+   Mailboxes are handled one at a time:
+
+   1. Go through all saved messages. If we've already seen an instance of this
+      message, try to copy it. Otherwise save a new instance of it.
+   2. Some of the copies may fail because they're already expunged by that
+      time. A list of these failed copies are saved to copy_retry_indexes.
+   3. UID conflicts are resolved by assigning a new UID to the message.
+      To avoid delays with remote dsync, this is done via worker API.
+      Internally the local worker copies the message to its new UID and
+      once the copy succeeds, the old UID is expunged. If the copy fails, it's
+      either due to message already being expunged or something more fatal.
+   4. Once all messages are saved/copied, see if there are any failed copies.
+      If so, goto 1, but going through only the failed messages.
+   5. If there are more mailboxes left, go to next one and goto 1.
+
+   Step 4 may require waiting for remote worker to send all replies.
+*/
+
+#include "lib.h"
+#include "array.h"
+#include "istream.h"
+#include "hash.h"
+#include "dsync-worker.h"
+#include "dsync-brain-private.h"
+
+struct dsync_brain_msg_copy_context {
+	struct dsync_brain_msg_iter *iter;
+	unsigned int msg_idx;
+};
+
+struct dsync_brain_msg_save_context {
+	struct dsync_brain_msg_iter *iter;
+	const struct dsync_message *msg;
+	unsigned int mailbox_idx;
+};
+
+static void
+dsync_brain_msg_sync_add_new_msgs(struct dsync_brain_msg_iter *iter);
+
+static void msg_save_callback(void *context)
+{
+	struct dsync_brain_msg_save_context *ctx = context;
+
+	if (--ctx->iter->save_results_left == 0 && !ctx->iter->adding_msgs)
+		dsync_brain_msg_sync_add_new_msgs(ctx->iter);
+	i_free(ctx);
+}
+
+static void msg_get_callback(enum dsync_msg_get_result result,
+			     const struct dsync_msg_static_data *data,
+			     void *context)
+{
+	struct dsync_brain_msg_save_context *ctx = context;
+	const struct dsync_brain_mailbox *mailbox;
+	struct istream *input;
+
+	i_assert(ctx->iter->save_results_left > 0);
+
+	mailbox = array_idx(&ctx->iter->sync->mailboxes, ctx->mailbox_idx);
+	switch (result) {
+	case DSYNC_MSG_GET_RESULT_SUCCESS:
+		/* the mailbox may have changed, make sure we've the
+		   right one */
+		dsync_worker_select_mailbox(ctx->iter->worker, &mailbox->box);
+
+		input = data->input;
+		dsync_worker_msg_save(ctx->iter->worker, ctx->msg, data,
+				      msg_save_callback, ctx);
+		i_stream_unref(&input);
+		break;
+	case DSYNC_MSG_GET_RESULT_EXPUNGED:
+		/* mail got expunged during sync. just skip this. */
+		msg_save_callback(ctx);
+		break;
+	case DSYNC_MSG_GET_RESULT_FAILED:
+		i_error("msg-get failed: box=%s uid=%u guid=%s",
+			mailbox->box.name, ctx->msg->uid, ctx->msg->guid);
+		dsync_brain_fail(ctx->iter->sync->brain);
+		msg_save_callback(ctx);
+		break;
+	}
+}
+
+static void
+dsync_brain_sync_remove_guid_instance(struct dsync_brain_msg_iter *iter,
+				      const struct dsync_brain_new_msg *msg)
+{
+	struct dsync_brain_guid_instance *inst;
+	void *orig_key, *orig_value;
+
+	if (!hash_table_lookup_full(iter->guid_hash, msg->msg->guid,
+				    &orig_key, &orig_value)) {
+		/* another failed copy already removed it */
+		return;
+	}
+	inst = orig_value;
+
+	if (inst->next == NULL)
+		hash_table_remove(iter->guid_hash, orig_key);
+	else
+		hash_table_update(iter->guid_hash, orig_key, inst->next);
+}
+
+static void dsync_brain_copy_callback(bool success, void *context)
+{
+	struct dsync_brain_msg_copy_context *ctx = context;
+	struct dsync_brain_new_msg *msg;
+
+	if (!success) {
+		/* mark the guid instance invalid and try again later */
+		msg = array_idx_modifiable(&ctx->iter->new_msgs, ctx->msg_idx);
+		i_assert(msg->saved);
+		msg->saved = FALSE;
+
+		if (ctx->iter->next_new_msg > ctx->msg_idx)
+			ctx->iter->next_new_msg = ctx->msg_idx;
+
+		dsync_brain_sync_remove_guid_instance(ctx->iter, msg);
+	}
+
+	if (--ctx->iter->copy_results_left == 0 && !ctx->iter->adding_msgs)
+		dsync_brain_msg_sync_add_new_msgs(ctx->iter);
+	i_free(ctx);
+}
+
+static int
+dsync_brain_msg_sync_add_new_msg(struct dsync_brain_msg_iter *dest_iter,
+				 const mailbox_guid_t *src_mailbox,
+				 unsigned int msg_idx,
+				 struct dsync_brain_new_msg *msg)
+{
+	struct dsync_brain_msg_save_context *save_ctx;
+	struct dsync_brain_msg_copy_context *copy_ctx;
+	struct dsync_brain_msg_iter *src_iter;
+	const struct dsync_brain_guid_instance *inst;
+	const struct dsync_brain_mailbox *inst_box;
+
+	msg->saved = TRUE;
+
+	inst = hash_table_lookup(dest_iter->guid_hash, msg->msg->guid);
+	if (inst != NULL) {
+		/* we can save this by copying an existing message */
+		inst_box = array_idx(&dest_iter->sync->mailboxes,
+				     inst->mailbox_idx);
+
+		copy_ctx = i_new(struct dsync_brain_msg_copy_context, 1);
+		copy_ctx->iter = dest_iter;
+		copy_ctx->msg_idx = msg_idx;
+
+		dest_iter->copy_results_left++;
+		dest_iter->adding_msgs = TRUE;
+		dsync_worker_msg_copy(dest_iter->worker,
+				      &inst_box->box.mailbox_guid,
+				      inst->uid, msg->msg,
+				      dsync_brain_copy_callback, copy_ctx);
+		dest_iter->adding_msgs = FALSE;
+	} else {
+		src_iter = dest_iter == dest_iter->sync->dest_msg_iter ?
+			dest_iter->sync->src_msg_iter :
+			dest_iter->sync->dest_msg_iter;
+
+		save_ctx = i_new(struct dsync_brain_msg_save_context, 1);
+		save_ctx->iter = dest_iter;
+		save_ctx->msg = msg->msg;
+		save_ctx->mailbox_idx = dest_iter->mailbox_idx;
+
+		dest_iter->save_results_left++;
+		dest_iter->adding_msgs = TRUE;
+		dsync_worker_msg_get(src_iter->worker, src_mailbox,
+				     msg->orig_uid, msg_get_callback, save_ctx);
+		dest_iter->adding_msgs = FALSE;
+		if (dsync_worker_output_flush(src_iter->worker) < 0)
+			return -1;
+		if (dsync_worker_is_output_full(dest_iter->worker)) {
+			/* see if the output becomes less full by flushing */
+			if (dsync_worker_output_flush(dest_iter->worker) < 0)
+				return -1;
+		}
+	}
+	return dsync_worker_is_output_full(dest_iter->worker) ? 0 : 1;
+}
+
+static bool
+dsync_brain_mailbox_add_new_msgs(struct dsync_brain_msg_iter *iter,
+				 const mailbox_guid_t *mailbox_guid)
+{
+	struct dsync_brain_new_msg *msgs;
+	unsigned int msg_count;
+	bool ret = TRUE;
+
+	msgs = array_get_modifiable(&iter->new_msgs, &msg_count);
+	while (iter->next_new_msg < msg_count) {
+		struct dsync_brain_new_msg *msg = &msgs[iter->next_new_msg];
+
+		if (msg->mailbox_idx != iter->mailbox_idx) {
+			i_assert(msg->mailbox_idx > iter->mailbox_idx);
+			ret = FALSE;
+			break;
+		}
+		iter->next_new_msg++;
+
+		if (msg->saved)
+			continue;
+		if (dsync_brain_msg_sync_add_new_msg(iter, mailbox_guid,
+						     iter->next_new_msg - 1,
+						     msg) <= 0) {
+			/* failed / continue later */
+			break;
+		}
+	}
+	if (iter->next_new_msg == msg_count)
+		ret = FALSE;
+
+	/* flush copy commands */
+	if (dsync_worker_output_flush(iter->worker) > 0 && ret) {
+		/* we have more space again, continue */
+		return dsync_brain_mailbox_add_new_msgs(iter, mailbox_guid);
+	} else {
+		return ret;
+	}
+}
+
+static void
+dsync_brain_mailbox_save_conflicts(struct dsync_brain_msg_iter *iter)
+{
+	const struct dsync_brain_uid_conflict *conflicts;
+	unsigned int i, count;
+
+	conflicts = array_get(&iter->uid_conflicts, &count);
+	for (i = iter->next_conflict; i < count; i++) {
+		if (conflicts[i].mailbox_idx != iter->mailbox_idx)
+			break;
+
+		dsync_worker_msg_update_uid(iter->worker, conflicts[i].old_uid,
+					    conflicts[i].new_uid);
+	}
+	iter->next_conflict = i;
+}
+
+static void
+dsync_brain_msg_sync_finish(struct dsync_brain_msg_iter *iter)
+{
+	struct dsync_brain_mailbox_sync *sync = iter->sync;
+
+	i_assert(sync->brain->state == DSYNC_STATE_SYNC_MSGS);
+
+	iter->msgs_sent = TRUE;
+
+	/* done with all mailboxes from this iter */
+	dsync_worker_set_input_callback(iter->worker, NULL, NULL);
+
+	if (sync->src_msg_iter->msgs_sent &&
+	    sync->dest_msg_iter->msgs_sent &&
+	    sync->src_msg_iter->save_results_left == 0 &&
+	    sync->dest_msg_iter->save_results_left == 0 &&
+	    dsync_worker_output_flush(sync->dest_worker) > 0 &&
+	    dsync_worker_output_flush(sync->src_worker) > 0) {
+		dsync_worker_set_output_callback(sync->src_msg_iter->worker,
+						 NULL, NULL);
+		dsync_worker_set_output_callback(sync->dest_msg_iter->worker,
+						 NULL, NULL);
+		sync->brain->state++;
+		dsync_brain_sync(sync->brain);
+	}
+}
+
+static bool
+dsync_brain_msg_sync_select_mailbox(struct dsync_brain_msg_iter *iter)
+{
+	const struct dsync_brain_mailbox *mailbox;
+
+	while (iter->mailbox_idx < array_count(&iter->sync->mailboxes)) {
+		if (array_count(&iter->new_msgs) == 0 &&
+		    array_count(&iter->uid_conflicts) == 0) {
+			/* optimization: don't even bother selecting this
+			   mailbox */
+			iter->mailbox_idx++;
+			continue;
+		}
+
+		mailbox = array_idx(&iter->sync->mailboxes, iter->mailbox_idx);
+		dsync_worker_select_mailbox(iter->worker, &mailbox->box);
+		return TRUE;
+	}
+	dsync_brain_msg_sync_finish(iter);
+	return FALSE;
+}
+
+static void
+dsync_brain_msg_sync_add_new_msgs(struct dsync_brain_msg_iter *iter)
+{
+	const struct dsync_brain_mailbox *mailbox;
+	const mailbox_guid_t *mailbox_guid;
+
+	if (iter->msgs_sent) {
+		dsync_brain_msg_sync_finish(iter);
+		return;
+	}
+
+	do {
+		mailbox = array_idx(&iter->sync->mailboxes, iter->mailbox_idx);
+		mailbox_guid = &mailbox->box.mailbox_guid;
+		if (dsync_brain_mailbox_add_new_msgs(iter, mailbox_guid)) {
+			/* continue later */
+			return;
+		}
+
+		/* all messages saved for this mailbox. continue with saving
+		   its conflicts and waiting for copies to finish. */
+		dsync_brain_mailbox_save_conflicts(iter);
+		if (iter->save_results_left > 0 ||
+		    iter->copy_results_left > 0) {
+			/* wait for saves/copies to finish */
+			return;
+		}
+
+		/* done with this mailbox, try the next one */
+		iter->mailbox_idx++;
+	} while (dsync_brain_msg_sync_select_mailbox(iter));
+}
+
+static void dsync_worker_new_msg_output(void *context)
+{
+	struct dsync_brain_msg_iter *iter = context;
+
+	dsync_brain_msg_sync_add_new_msgs(iter);
+}
+
+static int dsync_brain_new_msg_cmp(const struct dsync_brain_new_msg *m1,
+				   const struct dsync_brain_new_msg *m2)
+{
+	if (m1->mailbox_idx < m2->mailbox_idx)
+		return -1;
+	if (m1->mailbox_idx > m2->mailbox_idx)
+		return 1;
+
+	if (m1->msg->uid < m2->msg->uid)
+		return -1;
+	if (m1->msg->uid > m2->msg->uid)
+		return 1;
+	return 0;
+}
+
+static int
+dsync_brain_uid_conflict_cmp(const struct dsync_brain_uid_conflict *c1,
+			     const struct dsync_brain_uid_conflict *c2)
+{
+	if (c1->mailbox_idx < c2->mailbox_idx)
+	       return -1;
+	if (c1->mailbox_idx < c2->mailbox_idx)
+		return 1;
+
+	if (c1->new_uid < c2->new_uid)
+		return -1;
+	if (c1->new_uid > c2->new_uid)
+		return 1;
+	return 0;
+}
+
+static void
+dsync_brain_msg_iter_sync_new_msgs(struct dsync_brain_msg_iter *iter)
+{
+	iter->mailbox_idx = 0;
+
+	/* sort input by 1) mailbox, 2) new message UID */
+	array_sort(&iter->new_msgs, dsync_brain_new_msg_cmp);
+	array_sort(&iter->uid_conflicts, dsync_brain_uid_conflict_cmp);
+
+	dsync_worker_set_input_callback(iter->worker, NULL, iter);
+	dsync_worker_set_output_callback(iter->worker,
+					 dsync_worker_new_msg_output, iter);
+
+	if (dsync_brain_msg_sync_select_mailbox(iter))
+		dsync_brain_msg_sync_add_new_msgs(iter);
+}
+
+void dsync_brain_msg_sync_new_msgs(struct dsync_brain_mailbox_sync *sync)
+{
+	dsync_brain_msg_iter_sync_new_msgs(sync->src_msg_iter);
+	dsync_brain_msg_iter_sync_new_msgs(sync->dest_msg_iter);
+}

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/doveadm/dsync/dsync-brain-msgs.c	Thu Dec 29 14:43:45 2011 +0200
@@ -0,0 +1,537 @@
+/* Copyright (c) 2009-2011 Dovecot authors, see the included COPYING file */
+
+/* This code synchronizes messages in all mailboxes between two workers.
+   The "src" and "dest" terms don't really have anything to do with reality,
+   they're both treated equal.
+
+   1. Iterate through all messages in all (wanted) mailboxes. The mailboxes
+      are iterated in the same order and messages in ascending order.
+      All of the expunged messages at the end of mailbox (i.e.
+      last_existing_uid+1 .. next_uid-1) are also returned with
+      DSYNC_MAIL_FLAG_EXPUNGED set. We only care about the end of the mailbox,
+      because we can detect UID conflicts for messages in the middle by looking
+      at the next existing message and seeing if it has UID conflict.
+   2. For each seen non-expunged message, save it to GUID instance hash table:
+      message GUID => linked list of { uid, mailbox }
+   3. Each message in a mailbox is matched between the two workers as long as
+      both have messages left (the last ones may be expunged).
+      The possibilities are:
+
+      i) We don't know the GUIDs of both messages:
+
+	a) Message is expunged in both. Do nothing.
+	b) Message is expunged in only one of them. If there have been no UID
+	   conflicts seen so far, expunge the message in the other one.
+	   Otherwise, give the existing a message a new UID (at step 6).
+
+      ii) We know GUIDs of both messages (one/both of them may be expunged):
+
+	a) Messages have conflicting GUIDs. Give new UIDs for the non-expunged
+	   message(s) (at step 6).
+	b) Messages have matching GUIDs and one of them is expunged.
+	   Expunge also the other one. (We don't need to care about previous
+	   UID conflicts here, because we know this message is the same with
+	   both workers, since they have the same GUID.)
+	c) Messages have matching GUIDs and both of them exist. Sync flags from
+	   whichever has the higher modseq. If both modseqs equal but flags
+	   don't, pick the one that has more flags. If even the flag count is
+	   the same, just pick one of them.
+   4. One of the workers may messages left in the mailbox. Copy these
+      (non-expunged) messages to the other worker (at step 6).
+   5. If there are more mailboxes left, go to next one and goto 2.
+
+   6. Copy the new messages and give new UIDs to conflicting messages.
+      This code exists in dsync-brain-msgs-new.c
+*/
+
+#include "lib.h"
+#include "array.h"
+#include "hash.h"
+#include "dsync-worker.h"
+#include "dsync-brain-private.h"
+
+static void dsync_brain_guid_add(struct dsync_brain_msg_iter *iter)
+{
+	struct dsync_brain_guid_instance *inst, *prev_inst;
+
+	if ((iter->msg.flags & DSYNC_MAIL_FLAG_EXPUNGED) != 0)
+		return;
+
+	inst = p_new(iter->sync->pool, struct dsync_brain_guid_instance, 1);
+	inst->mailbox_idx = iter->mailbox_idx;
+	inst->uid = iter->msg.uid;
+
+	prev_inst = hash_table_lookup(iter->guid_hash, iter->msg.guid);
+	if (prev_inst == NULL) {
+		hash_table_insert(iter->guid_hash,
+				  p_strdup(iter->sync->pool, iter->msg.guid),
+				  inst);
+	} else {
+		inst->next = prev_inst->next;
+		prev_inst->next = inst;
+	}
+}
+
+static int dsync_brain_msg_iter_next(struct dsync_brain_msg_iter *iter)
+{
+	int ret = 1;
+
+	if (iter->msg.guid == NULL) {
+		ret = dsync_worker_msg_iter_next(iter->iter,
+						 &iter->mailbox_idx,
+						 &iter->msg);
+		if (ret > 0)
+			dsync_brain_guid_add(iter);
+	}
+
+	if (iter->sync->wanted_mailbox_idx != iter->mailbox_idx) {
+		/* finished with this mailbox */
+		return -1;
+	}
+	return ret;
+}
+
+static int
+dsync_brain_msg_iter_skip_mailbox(struct dsync_brain_mailbox_sync *sync)
+{
+	int ret;
+
+	while ((ret = dsync_brain_msg_iter_next(sync->src_msg_iter)) > 0)
+		sync->src_msg_iter->msg.guid = NULL;
+	if (ret == 0)
+		return 0;
+
+	while ((ret = dsync_brain_msg_iter_next(sync->dest_msg_iter)) > 0)
+		sync->dest_msg_iter->msg.guid = NULL;
+	if (ret == 0)
+		return 0;
+
+	sync->skip_mailbox = FALSE;
+	return -1;
+}
+
+static int dsync_brain_msg_iter_next_pair(struct dsync_brain_mailbox_sync *sync)
+{
+	int ret1, ret2;
+
+	if (sync->skip_mailbox) {
+		if (dsync_brain_msg_iter_skip_mailbox(sync) == 0)
+			return 0;
+	}
+
+	ret1 = dsync_brain_msg_iter_next(sync->src_msg_iter);
+	ret2 = dsync_brain_msg_iter_next(sync->dest_msg_iter);
+	if (ret1 == 0 || ret2 == 0) {
+		/* make sure we iterate through everything in both iterators
+		   (even if it might not seem necessary, because proxy
+		   requires it) */
+		return 0;
+	}
+	if (ret1 < 0 || ret2 < 0)
+		return -1;
+	return 1;
+}
+
+static void
+dsync_brain_msg_sync_save(struct dsync_brain_msg_iter *iter,
+			  unsigned int mailbox_idx,
+			  const struct dsync_message *msg)
+{
+	struct dsync_brain_new_msg *new_msg;
+
+	if ((msg->flags & DSYNC_MAIL_FLAG_EXPUNGED) != 0)
+		return;
+
+	new_msg = array_append_space(&iter->new_msgs);
+	new_msg->mailbox_idx = mailbox_idx;
+	new_msg->orig_uid = msg->uid;
+	new_msg->msg = dsync_message_dup(iter->sync->pool, msg);
+}
+
+static void
+dsync_brain_msg_sync_conflict(struct dsync_brain_msg_iter *conflict_iter,
+			      struct dsync_brain_msg_iter *save_iter,
+			      const struct dsync_message *msg)
+{
+	struct dsync_brain_uid_conflict *conflict;
+	struct dsync_brain_new_msg *new_msg;
+	struct dsync_brain_mailbox *brain_box;
+	uint32_t new_uid;
+
+	brain_box = array_idx_modifiable(&save_iter->sync->mailboxes,
+					 save_iter->mailbox_idx);
+
+	if (save_iter->sync->brain->backup) {
+		i_warning("Destination mailbox %s has been modified, "
+			  "need to recreate it before we can continue syncing",
+			  brain_box->box.name);
+		dsync_worker_delete_mailbox(save_iter->sync->brain->dest_worker,
+					    &brain_box->box);
+		save_iter->sync->brain->unexpected_changes = TRUE;
+		save_iter->sync->skip_mailbox = TRUE;
+		return;
+	}
+
+	new_uid = brain_box->box.uid_next++;
+
+	conflict = array_append_space(&conflict_iter->uid_conflicts);
+	conflict->mailbox_idx = conflict_iter->mailbox_idx;
+	conflict->old_uid = msg->uid;
+	conflict->new_uid = new_uid;
+
+	new_msg = array_append_space(&save_iter->new_msgs);
+	new_msg->mailbox_idx = save_iter->mailbox_idx;
+	new_msg->orig_uid = msg->uid;
+	new_msg->msg = dsync_message_dup(save_iter->sync->pool, msg);
+	new_msg->msg->uid = new_uid;
+}
+
+static int
+dsync_message_flag_importance_cmp(const struct dsync_message *m1,
+				  const struct dsync_message *m2)
+{
+	unsigned int i, count1, count2;
+
+	if (m1->modseq > m2->modseq)
+		return -1;
+	else if (m1->modseq < m2->modseq)
+		return 1;
+
+	if (m1->flags == m2->flags &&
+	    dsync_keyword_list_equals(m1->keywords, m2->keywords))
+		return 0;
+
+	/* modseqs match, but flags aren't the same. pick the one that
+	   has more flags. */
+	count1 = str_array_length(m1->keywords);
+	count2 = str_array_length(m2->keywords);
+	for (i = 1; i != MAIL_RECENT; i <<= 1) {
+		if ((m1->flags & i) != 0)
+			count1++;
+		if ((m2->flags & i) != 0)
+			count2++;
+	}
+	if (count1 > count2)
+		return -1;
+	else if (count1 < count2)
+		return 1;
+
+	/* they even have the same number of flags. don't bother with further
+	   guessing, just pick the first one. */
+	return -1;
+}
+
+static void dsync_brain_msg_sync_existing(struct dsync_brain_mailbox_sync *sync,
+					  struct dsync_message *src_msg,
+					  struct dsync_message *dest_msg)
+{
+	int ret;
+
+	ret = dsync_message_flag_importance_cmp(src_msg, dest_msg);
+	if (ret < 0 || (sync->brain->backup && ret > 0))
+		dsync_worker_msg_update_metadata(sync->dest_worker, src_msg);
+	else if (ret > 0)
+		dsync_worker_msg_update_metadata(sync->src_worker, dest_msg);
+}
+
+static int dsync_brain_msg_sync_pair(struct dsync_brain_mailbox_sync *sync)
+{
+	struct dsync_message *src_msg = &sync->src_msg_iter->msg;
+	struct dsync_message *dest_msg = &sync->dest_msg_iter->msg;
+	const char *src_guid, *dest_guid;
+	unsigned char guid_128_data[GUID_128_SIZE * 2 + 1];
+	bool src_expunged, dest_expunged;
+
+	i_assert(sync->src_msg_iter->mailbox_idx ==
+		 sync->dest_msg_iter->mailbox_idx);
+
+	src_expunged = (src_msg->flags & DSYNC_MAIL_FLAG_EXPUNGED) != 0;
+	dest_expunged = (dest_msg->flags & DSYNC_MAIL_FLAG_EXPUNGED) != 0;
+
+	/* If a message is expunged, it's guaranteed to have a 128bit GUID.
+	   If the other message isn't expunged, we'll need to convert its GUID
+	   to the 128bit GUID form (if it's not already) so that we can compare
+	   them. */
+	if (src_expunged) {
+		src_guid = src_msg->guid;
+		dest_guid = dsync_get_guid_128_str(dest_msg->guid,
+						   guid_128_data,
+						   sizeof(guid_128_data));
+	} else if (dest_expunged) {
+		src_guid = dsync_get_guid_128_str(src_msg->guid, guid_128_data,
+						  sizeof(guid_128_data));
+		dest_guid = dest_msg->guid;
+	} else {
+		src_guid = src_msg->guid;
+		dest_guid = dest_msg->guid;
+	}
+
+	/* FIXME: checking for sync->uid_conflict isn't fully reliable here.
+	   we should be checking if the next matching message pair has a
+	   conflict, not if the previous pair had one. */
+	if (src_msg->uid < dest_msg->uid) {
+		/* message has been expunged from dest. */
+		if (src_expunged) {
+			/* expunged from source already */
+		} else if (sync->uid_conflict || sync->brain->backup) {
+			/* update uid src, copy to dest */
+			dsync_brain_msg_sync_conflict(sync->src_msg_iter,
+						      sync->dest_msg_iter,
+						      src_msg);
+		} else {
+			/* expunge from source */
+			dsync_worker_msg_expunge(sync->src_worker,
+						 src_msg->uid);
+		}
+		src_msg->guid = NULL;
+		return 0;
+	} else if (src_msg->uid > dest_msg->uid) {
+		/* message has been expunged from src. */
+		if (dest_expunged) {
+			/* expunged from dest already */
+		} else if (sync->uid_conflict && !sync->brain->backup) {
+			/* update uid in dest, copy to src */
+			dsync_brain_msg_sync_conflict(sync->dest_msg_iter,
+						      sync->src_msg_iter,
+						      dest_msg);
+		} else {
+			/* expunge from dest */
+			dsync_worker_msg_expunge(sync->dest_worker,
+						 dest_msg->uid);
+		}
+		dest_msg->guid = NULL;
+		return 0;
+	}
+
+	/* UIDs match, but do GUIDs? If either of the GUIDs aren't set, it
+	   means that either the storage doesn't support GUIDs or we're
+	   handling an old-style expunge record. In that case just assume
+	   they match. */
+	if (strcmp(src_guid, dest_guid) != 0 &&
+	    *src_guid != '\0' && *dest_guid != '\0') {
+		/* UID conflict. give new UIDs to messages in both src and
+		   dest (if they're not expunged already) */
+		sync->uid_conflict = TRUE;
+		if (!dest_expunged) {
+			dsync_brain_msg_sync_conflict(sync->dest_msg_iter,
+						      sync->src_msg_iter,
+						      dest_msg);
+		}
+		if (!src_expunged) {
+			dsync_brain_msg_sync_conflict(sync->src_msg_iter,
+						      sync->dest_msg_iter,
+						      src_msg);
+		}
+	} else if (dest_expunged) {
+		/* message expunged from destination */
+		if (src_expunged) {
+			/* expunged from source already */
+		} else if (sync->brain->backup) {
+			dsync_brain_msg_sync_conflict(sync->src_msg_iter,
+						      sync->dest_msg_iter,
+						      src_msg);
+		} else {
+			dsync_worker_msg_expunge(sync->src_worker,
+						 src_msg->uid);
+		}
+	} else if (src_expunged) {
+		/* message expunged from source, expunge from destination too */
+		dsync_worker_msg_expunge(sync->dest_worker, dest_msg->uid);
+	} else {
+		/* message exists in both source and dest, sync metadata */
+		dsync_brain_msg_sync_existing(sync, src_msg, dest_msg);
+	}
+	src_msg->guid = NULL;
+	dest_msg->guid = NULL;
+	return 0;
+}
+
+static bool dsync_brain_msg_sync_mailbox_end(struct dsync_brain_msg_iter *iter1,
+					     struct dsync_brain_msg_iter *iter2)
+{
+	int ret;
+
+	while ((ret = dsync_brain_msg_iter_next(iter1)) > 0) {
+		dsync_brain_msg_sync_save(iter2, iter1->mailbox_idx,
+					  &iter1->msg);
+		iter1->msg.guid = NULL;
+	}
+	return ret < 0;
+}
+
+static bool
+dsync_brain_msg_sync_mailbox_more(struct dsync_brain_mailbox_sync *sync)
+{
+	int ret;
+
+	while ((ret = dsync_brain_msg_iter_next_pair(sync)) > 0) {
+		if (dsync_brain_msg_sync_pair(sync) < 0)
+			break;
+		if (dsync_worker_is_output_full(sync->dest_worker)) {
+			if (dsync_worker_output_flush(sync->dest_worker) <= 0)
+				return FALSE;
+		}
+	}
+ 	if (ret == 0)
+		return FALSE;
+
+	/* finished syncing messages in this mailbox that exist in both source
+	   and destination. if there are messages left, we can't reliably know
+	   if they should be expunged, so just copy them to the other side. */
+	if (!sync->brain->backup) {
+		if (!dsync_brain_msg_sync_mailbox_end(sync->dest_msg_iter,
+						      sync->src_msg_iter))
+			return FALSE;
+	}
+	if (!dsync_brain_msg_sync_mailbox_end(sync->src_msg_iter,
+					      sync->dest_msg_iter))
+		return FALSE;
+
+	/* done with this mailbox. the same iterator is still used for
+	   getting messages from other mailboxes. */
+	return TRUE;
+}
+
+void dsync_brain_msg_sync_more(struct dsync_brain_mailbox_sync *sync)
+{
+	const struct dsync_brain_mailbox *mailboxes;
+	unsigned int count, mailbox_idx = 0;
+
+	mailboxes = array_get(&sync->mailboxes, &count);
+	while (dsync_brain_msg_sync_mailbox_more(sync)) {
+		/* sync the next mailbox */
+		sync->uid_conflict = FALSE;
+		mailbox_idx = ++sync->wanted_mailbox_idx;
+		if (mailbox_idx >= count)
+			break;
+
+		dsync_worker_select_mailbox(sync->src_worker,
+			&mailboxes[mailbox_idx].box);
+		dsync_worker_select_mailbox(sync->dest_worker,
+			&mailboxes[mailbox_idx].box);
+	}
+	if (mailbox_idx < count) {
+		/* output buffer is full */
+		return;
+	}
+
+	/* finished with all mailboxes. */
+	dsync_worker_set_input_callback(sync->src_msg_iter->worker, NULL, NULL);
+	dsync_worker_set_output_callback(sync->src_msg_iter->worker, NULL, NULL);
+	dsync_worker_set_input_callback(sync->dest_msg_iter->worker, NULL, NULL);
+	dsync_worker_set_output_callback(sync->dest_msg_iter->worker, NULL, NULL);
+
+	if (dsync_worker_msg_iter_deinit(&sync->src_msg_iter->iter) < 0 ||
+	    dsync_worker_msg_iter_deinit(&sync->dest_msg_iter->iter) < 0) {
+		dsync_brain_fail(sync->brain);
+		return;
+	}
+
+	dsync_brain_msg_sync_new_msgs(sync);
+}
+
+static void dsync_worker_msg_callback(void *context)
+{
+	struct dsync_brain_mailbox_sync *sync = context;
+
+	dsync_brain_msg_sync_more(sync);
+}
+
+static struct dsync_brain_msg_iter *
+dsync_brain_msg_iter_init(struct dsync_brain_mailbox_sync *sync,
+			  struct dsync_worker *worker,
+			  const mailbox_guid_t mailboxes[],
+			  unsigned int mailbox_count)
+{
+	struct dsync_brain_msg_iter *iter;
+
+	iter = p_new(sync->pool, struct dsync_brain_msg_iter, 1);
+	iter->sync = sync;
+	iter->worker = worker;
+	i_array_init(&iter->uid_conflicts, 128);
+	i_array_init(&iter->new_msgs, 128);
+	iter->guid_hash = hash_table_create(default_pool, sync->pool, 10000,
+					    strcase_hash,
+					    (hash_cmp_callback_t *)strcasecmp);
+
+	iter->iter = dsync_worker_msg_iter_init(worker, mailboxes,
+						mailbox_count);
+	dsync_worker_set_input_callback(worker,
+					dsync_worker_msg_callback, sync);
+	dsync_worker_set_output_callback(worker,
+					 dsync_worker_msg_callback, sync);
+	if (mailbox_count > 0) {
+		const struct dsync_brain_mailbox *first;
+
+		first = array_idx(&sync->mailboxes, 0);
+		dsync_worker_select_mailbox(worker, &first->box);
+	}
+	return iter;
+}
+
+static void dsync_brain_msg_iter_deinit(struct dsync_brain_msg_iter *iter)
+{
+	if (iter->iter != NULL)
+		(void)dsync_worker_msg_iter_deinit(&iter->iter);
+
+	hash_table_destroy(&iter->guid_hash);
+	array_free(&iter->uid_conflicts);
+	array_free(&iter->new_msgs);
+}
+
+static void
+get_mailbox_guids(const ARRAY_TYPE(dsync_brain_mailbox) *mailboxes,
+		  ARRAY_TYPE(mailbox_guid) *guids)
+{
+	const struct dsync_brain_mailbox *brain_box;
+
+	t_array_init(guids, array_count(mailboxes));
+	array_foreach(mailboxes, brain_box)
+		array_append(guids, &brain_box->box.mailbox_guid, 1);
+}
+
+struct dsync_brain_mailbox_sync *
+dsync_brain_msg_sync_init(struct dsync_brain *brain,
+			  const ARRAY_TYPE(dsync_brain_mailbox) *mailboxes)
+{
+	struct dsync_brain_mailbox_sync *sync;
+	pool_t pool;
+
+	pool = pool_alloconly_create("dsync brain mailbox sync", 1024*256);
+	sync = p_new(pool, struct dsync_brain_mailbox_sync, 1);
+	sync->pool = pool;
+	sync->brain = brain;
+	sync->src_worker = brain->src_worker;
+	sync->dest_worker = brain->dest_worker;
+
+	p_array_init(&sync->mailboxes, pool, array_count(mailboxes));
+	array_append_array(&sync->mailboxes, mailboxes);
+	T_BEGIN {
+		ARRAY_TYPE(mailbox_guid) guids_arr;
+		const mailbox_guid_t *guids;
+		unsigned int count;
+
+		get_mailbox_guids(mailboxes, &guids_arr);
+
+		/* initialize message iteration on both workers */
+		guids = array_get(&guids_arr, &count);
+		sync->src_msg_iter =
+			dsync_brain_msg_iter_init(sync, brain->src_worker,
+						  guids, count);
+		sync->dest_msg_iter =
+			dsync_brain_msg_iter_init(sync, brain->dest_worker,
+						  guids, count);
+	} T_END;
+	return sync;
+}
+
+void dsync_brain_msg_sync_deinit(struct dsync_brain_mailbox_sync **_sync)
+{
+	struct dsync_brain_mailbox_sync *sync = *_sync;
+
+	*_sync = NULL;
+
+	dsync_brain_msg_iter_deinit(sync->src_msg_iter);
+	dsync_brain_msg_iter_deinit(sync->dest_msg_iter);
+	pool_unref(&sync->pool);
+}

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/doveadm/dsync/dsync-brain-private.h	Thu Dec 29 14:43:45 2011 +0200
@@ -0,0 +1,144 @@
+#ifndef DSYNC_BRAIN_PRIVATE_H
+#define DSYNC_BRAIN_PRIVATE_H
+
+#include "dsync-data.h"
+#include "dsync-brain.h"
+
+enum dsync_state {
+	DSYNC_STATE_GET_MAILBOXES = 0,
+	DSYNC_STATE_GET_SUBSCRIPTIONS,
+	DSYNC_STATE_SYNC_MAILBOXES,
+	DSYNC_STATE_SYNC_SUBSCRIPTIONS,
+	DSYNC_STATE_SYNC_MSGS,
+	DSYNC_STATE_SYNC_MSGS_FLUSH,
+	DSYNC_STATE_SYNC_MSGS_FLUSH2,
+	DSYNC_STATE_SYNC_UPDATE_MAILBOXES,
+	DSYNC_STATE_SYNC_FLUSH,
+	DSYNC_STATE_SYNC_FLUSH2,
+	DSYNC_STATE_SYNC_END
+};
+
+struct dsync_brain_mailbox_list {
+	pool_t pool;
+	struct dsync_brain *brain;
+	struct dsync_worker *worker;
+	struct dsync_worker_mailbox_iter *iter;
+	ARRAY_TYPE(dsync_mailbox) mailboxes;
+	ARRAY_TYPE(dsync_mailbox) dirs;
+};
+
+struct dsync_brain_subs_list {
+	pool_t pool;
+	struct dsync_brain *brain;
+	struct dsync_worker *worker;
+	struct dsync_worker_subs_iter *iter;
+	ARRAY_DEFINE(subscriptions, struct dsync_worker_subscription);
+	ARRAY_DEFINE(unsubscriptions, struct dsync_worker_unsubscription);
+};
+
+struct dsync_brain_guid_instance {
+	struct dsync_brain_guid_instance *next;
+	uint32_t uid;
+	/* mailbox index in dsync_brain_mailbox_list.mailboxes */
+	unsigned int mailbox_idx:31;
+	unsigned int failed:1;
+};
+
+struct dsync_brain_msg_iter {
+	struct dsync_brain_mailbox_sync *sync;
+	struct dsync_worker *worker;
+
+	struct dsync_worker_msg_iter *iter;
+	struct dsync_message msg;
+
+	unsigned int mailbox_idx;
+
+	/* char *guid -> struct dsync_brain_guid_instance* */
+	struct hash_table *guid_hash;
+
+	ARRAY_DEFINE(new_msgs, struct dsync_brain_new_msg);
+	ARRAY_DEFINE(uid_conflicts, struct dsync_brain_uid_conflict);
+	unsigned int next_new_msg, next_conflict;
+
+	/* copy operations that failed. indexes point to new_msgs array */
+	unsigned int copy_results_left;
+	unsigned int save_results_left;
+
+	unsigned int msgs_sent:1;
+	unsigned int adding_msgs:1;
+};
+
+struct dsync_brain_uid_conflict {
+	uint32_t mailbox_idx;
+	uint32_t old_uid, new_uid;
+};
+
+struct dsync_brain_new_msg {
+	unsigned int mailbox_idx:30;
+	/* TRUE if it currently looks like message has been saved/copied.
+	   if copying fails, it sets this back to FALSE and updates
+	   iter->next_new_msg. */
+	unsigned int saved:1;
+	uint32_t orig_uid;
+	struct dsync_message *msg;
+};
+
+struct dsync_brain_mailbox {
+	struct dsync_mailbox box;
+	struct dsync_mailbox *src;
+	struct dsync_mailbox *dest;
+};
+ARRAY_DEFINE_TYPE(dsync_brain_mailbox, struct dsync_brain_mailbox);
+
+struct dsync_brain_mailbox_sync {
+	struct dsync_brain *brain;
+	pool_t pool;
+
+	ARRAY_TYPE(dsync_brain_mailbox) mailboxes;
+	unsigned int wanted_mailbox_idx;
+
+	struct dsync_worker *src_worker;
+	struct dsync_worker *dest_worker;
+
+	struct dsync_brain_msg_iter *src_msg_iter;
+	struct dsync_brain_msg_iter *dest_msg_iter;
+
+	unsigned int uid_conflict:1;
+	unsigned int skip_mailbox:1;
+};
+
+struct dsync_brain {
+	struct dsync_worker *src_worker;
+	struct dsync_worker *dest_worker;
+	char *mailbox;
+	enum dsync_brain_flags flags;
+
+	enum dsync_state state;
+
+	struct dsync_brain_mailbox_list *src_mailbox_list;
+	struct dsync_brain_mailbox_list *dest_mailbox_list;
+
+	struct dsync_brain_subs_list *src_subs_list;
+	struct dsync_brain_subs_list *dest_subs_list;
+
+	struct dsync_brain_mailbox_sync *mailbox_sync;
+	struct timeout *to;
+
+	unsigned int failed:1;
+	unsigned int verbose:1;
+	unsigned int backup:1;
+	unsigned int unexpected_changes:1;
+	unsigned int stdout_tty:1;
+};
+
+void dsync_brain_fail(struct dsync_brain *brain);
+
+struct dsync_brain_mailbox_sync *
+dsync_brain_msg_sync_init(struct dsync_brain *brain,
+			  const ARRAY_TYPE(dsync_brain_mailbox) *mailboxes);
+void dsync_brain_msg_sync_more(struct dsync_brain_mailbox_sync *sync);
+void dsync_brain_msg_sync_deinit(struct dsync_brain_mailbox_sync **_sync);
+
+void dsync_brain_msg_sync_new_msgs(struct dsync_brain_mailbox_sync *sync);
+
+#endif

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/doveadm/dsync/dsync-brain.c	Thu Dec 29 14:43:45 2011 +0200
@@ -0,0 +1,917 @@
+/* Copyright (c) 2009-2011 Dovecot authors, see the included COPYING file */
+
+#include "lib.h"
+#include "array.h"
+#include "hash.h"
+#include "dsync-worker.h"
+#include "dsync-brain-private.h"
+
+#include <unistd.h>
+
+#define DSYNC_WRONG_DIRECTION_ERROR_MSG \
+	"dsync backup: " \
+	"Looks like you're trying to run backup in wrong direction. " \
+	"Source is empty and destination is not."
+
+static void
+dsync_brain_mailbox_list_deinit(struct dsync_brain_mailbox_list **list);
+static void
+dsync_brain_subs_list_deinit(struct dsync_brain_subs_list **list);
+
+struct dsync_brain *
+dsync_brain_init(struct dsync_worker *src_worker,
+		 struct dsync_worker *dest_worker,
+		 const char *mailbox, enum dsync_brain_flags flags)
+{
+	struct dsync_brain *brain;
+
+	brain = i_new(struct dsync_brain, 1);
+	brain->src_worker = src_worker;
+	brain->dest_worker = dest_worker;
+	brain->mailbox = i_strdup(mailbox);
+	brain->flags = flags;
+	brain->verbose = (flags & DSYNC_BRAIN_FLAG_VERBOSE) != 0;
+	brain->backup = (flags & DSYNC_BRAIN_FLAG_BACKUP) != 0;
+	brain->stdout_tty = isatty(STDOUT_FILENO) > 0;
+
+	if ((flags & DSYNC_BRAIN_FLAG_VERBOSE) != 0) {
+		dsync_worker_set_verbose(src_worker);
+		dsync_worker_set_verbose(dest_worker);
+	}
+	return brain;
+}
+
+void dsync_brain_fail(struct dsync_brain *brain)
+{
+	brain->failed = TRUE;
+	io_loop_stop(current_ioloop);
+}
+
+int dsync_brain_deinit(struct dsync_brain **_brain)
+{
+	struct dsync_brain *brain = *_brain;
+	int ret = brain->failed ? -1 : 0;
+
+	if (brain->state != DSYNC_STATE_SYNC_END)
+		ret = -1;
+	if (brain->to != NULL)
+		timeout_remove(&brain->to);
+
+	if (ret < 0) {
+		/* make sure we unreference save input streams before workers
+		   are deinitialized, so they can destroy the streams */
+		dsync_worker_msg_save_cancel(brain->src_worker);
+		dsync_worker_msg_save_cancel(brain->dest_worker);
+	}
+
+	if (brain->mailbox_sync != NULL)
+		dsync_brain_msg_sync_deinit(&brain->mailbox_sync);
+
+	if (brain->src_mailbox_list != NULL)
+		dsync_brain_mailbox_list_deinit(&brain->src_mailbox_list);
+	if (brain->dest_mailbox_list != NULL)
+		dsync_brain_mailbox_list_deinit(&brain->dest_mailbox_list);
+
+	if (brain->src_subs_list != NULL)
+		dsync_brain_subs_list_deinit(&brain->src_subs_list);
+	if (brain->dest_subs_list != NULL)
+		dsync_brain_subs_list_deinit(&brain->dest_subs_list);
+
+	if (dsync_worker_has_failed(brain->src_worker) ||
+	    dsync_worker_has_failed(brain->dest_worker))
+		ret = -1;
+
+	*_brain = NULL;
+	i_free(brain->mailbox);
+	i_free(brain);
+	return ret;
+}
+
+static void dsync_brain_mailbox_list_finished(struct dsync_brain *brain)
+{
+	if (brain->src_mailbox_list->iter != NULL ||
+	    brain->dest_mailbox_list->iter != NULL)
+		return;
+
+	/* both lists are finished */
+	brain->state++;
+	dsync_brain_sync(brain);
+}
+
+static void dsync_worker_mailbox_input(void *context)
+{
+	struct dsync_brain_mailbox_list *list = context;
+	struct dsync_mailbox dsync_box, *dup_box;
+	int ret;
+
+	while ((ret = dsync_worker_mailbox_iter_next(list->iter,
+						     &dsync_box)) > 0) {
+		if (list->brain->mailbox != NULL &&
+		    strcmp(list->brain->mailbox, dsync_box.name) != 0)
+			continue;
+
+		dup_box = dsync_mailbox_dup(list->pool, &dsync_box);
+		if (!dsync_mailbox_is_noselect(dup_box))
+			array_append(&list->mailboxes, &dup_box, 1);
+		else
+			array_append(&list->dirs, &dup_box, 1);
+	}
+	if (ret < 0) {
+		/* finished listing mailboxes */
+		if (dsync_worker_mailbox_iter_deinit(&list->iter) < 0)
+			dsync_brain_fail(list->brain);
+		array_sort(&list->mailboxes, dsync_mailbox_p_guid_cmp);
+		array_sort(&list->dirs, dsync_mailbox_p_name_sha1_cmp);
+		dsync_brain_mailbox_list_finished(list->brain);
+	}
+}
+
+static struct dsync_brain_mailbox_list *
+dsync_brain_mailbox_list_init(struct dsync_brain *brain,
+			      struct dsync_worker *worker)
+{
+	struct dsync_brain_mailbox_list *list;
+	pool_t pool;
+
+	pool = pool_alloconly_create("dsync brain mailbox list", 10240);
+	list = p_new(pool, struct dsync_brain_mailbox_list, 1);
+	list->pool = pool;
+	list->brain = brain;
+	list->worker = worker;
+	list->iter = dsync_worker_mailbox_iter_init(worker);
+	p_array_init(&list->mailboxes, pool, 128);
+	p_array_init(&list->dirs, pool, 32);
+	dsync_worker_set_input_callback(worker, dsync_worker_mailbox_input,
+					list);
+	return list;
+}
+
+static void
+dsync_brain_mailbox_list_deinit(struct dsync_brain_mailbox_list **_list)
+{
+	struct dsync_brain_mailbox_list *list = *_list;
+
+	*_list = NULL;
+
+	if (list->iter != NULL)
+		(void)dsync_worker_mailbox_iter_deinit(&list->iter);
+	pool_unref(&list->pool);
+}
+
+static void dsync_brain_subs_list_finished(struct dsync_brain *brain)
+{
+	if (brain->src_subs_list->iter != NULL ||
+	    brain->dest_subs_list->iter != NULL)
+		return;
+
+	/* both lists are finished */
+	brain->state++;
+	dsync_brain_sync(brain);
+}
+
+static int
+dsync_worker_subscription_cmp(const struct dsync_worker_subscription *s1,
+			      const struct dsync_worker_subscription *s2)
+{
+	return strcmp(s1->vname, s2->vname);
+}
+
+static int
+dsync_worker_unsubscription_cmp(const struct dsync_worker_unsubscription *u1,
+				const struct dsync_worker_unsubscription *u2)
+{
+	int ret;
+
+	ret = strcmp(u1->ns_prefix, u2->ns_prefix);
+	return ret != 0 ? ret :
+		dsync_guid_cmp(&u1->name_sha1, &u2->name_sha1);
+}
+
+static void dsync_worker_subs_input(void *context)
+{
+	struct dsync_brain_subs_list *list = context;
+	struct dsync_worker_subscription subs;
+	struct dsync_worker_unsubscription unsubs;
+	int ret;
+
+	memset(&subs, 0, sizeof(subs));
+	while ((ret = dsync_worker_subs_iter_next(list->iter, &subs)) > 0) {
+		subs.vname = p_strdup(list->pool, subs.vname);
+		subs.storage_name = p_strdup(list->pool, subs.storage_name);
+		subs.ns_prefix = p_strdup(list->pool, subs.ns_prefix);
+		array_append(&list->subscriptions, &subs, 1);
+	}
+	if (ret == 0)
+		return;
+
+	memset(&unsubs, 0, sizeof(unsubs));
+	while ((ret = dsync_worker_subs_iter_next_un(list->iter,
+						     &unsubs)) > 0) {
+		unsubs.ns_prefix = p_strdup(list->pool, unsubs.ns_prefix);
+		array_append(&list->unsubscriptions, &unsubs, 1);
+	}
+
+	if (ret < 0) {
+		/* finished listing subscriptions */
+		if (dsync_worker_subs_iter_deinit(&list->iter) < 0)
+			dsync_brain_fail(list->brain);
+		array_sort(&list->subscriptions,
+			   dsync_worker_subscription_cmp);
+		array_sort(&list->unsubscriptions,
+			   dsync_worker_unsubscription_cmp);
+		dsync_brain_subs_list_finished(list->brain);
+	}
+}
+
+static struct dsync_brain_subs_list *
+dsync_brain_subs_list_init(struct dsync_brain *brain,
+			      struct dsync_worker *worker)
+{
+	struct dsync_brain_subs_list *list;
+	pool_t pool;
+
+	pool = pool_alloconly_create("dsync brain subs list", 1024*4);
+	list = p_new(pool, struct dsync_brain_subs_list, 1);
+	list->pool = pool;
+	list->brain = brain;
+	list->worker = worker;
+	list->iter = dsync_worker_subs_iter_init(worker);
+	p_array_init(&list->subscriptions, pool, 128);
+	p_array_init(&list->unsubscriptions, pool, 64);
+	dsync_worker_set_input_callback(worker, dsync_worker_subs_input, list);
+	return list;
+}
+
+static void
+dsync_brain_subs_list_deinit(struct dsync_brain_subs_list **_list)
+{
+	struct dsync_brain_subs_list *list = *_list;
+
+	*_list = NULL;
+
+	if (list->iter != NULL)
+		(void)dsync_worker_subs_iter_deinit(&list->iter);
+	pool_unref(&list->pool);
+}
+
+enum dsync_brain_mailbox_action {
+	DSYNC_BRAIN_MAILBOX_ACTION_NONE,
+	DSYNC_BRAIN_MAILBOX_ACTION_CREATE,
+	DSYNC_BRAIN_MAILBOX_ACTION_DELETE
+};
+
+static void
+dsync_brain_mailbox_action(struct dsync_brain *brain,
+			   enum dsync_brain_mailbox_action action,
+			   struct dsync_worker *action_worker,
+			   struct dsync_mailbox *action_box)
+{
+	struct dsync_mailbox new_box;
+
+	if (brain->backup && action_worker == brain->src_worker) {
+		/* backup mode: switch actions */
+		action_worker = brain->dest_worker;
+		switch (action) {
+		case DSYNC_BRAIN_MAILBOX_ACTION_NONE:
+			break;
+		case DSYNC_BRAIN_MAILBOX_ACTION_CREATE:
+			action = DSYNC_BRAIN_MAILBOX_ACTION_DELETE;
+			break;
+		case DSYNC_BRAIN_MAILBOX_ACTION_DELETE:
+			action = DSYNC_BRAIN_MAILBOX_ACTION_CREATE;
+			break;
+		}
+	}
+
+	switch (action) {
+	case DSYNC_BRAIN_MAILBOX_ACTION_NONE:
+		break;
+	case DSYNC_BRAIN_MAILBOX_ACTION_CREATE:
+		new_box = *action_box;
+		new_box.uid_next = action_box->uid_validity == 0 ? 0 : 1;
+		new_box.first_recent_uid = 0;
+		new_box.highest_modseq = 0;
+		dsync_worker_create_mailbox(action_worker, &new_box);
+		break;
+	case DSYNC_BRAIN_MAILBOX_ACTION_DELETE:
+		if (!dsync_mailbox_is_noselect(action_box))
+			dsync_worker_delete_mailbox(action_worker, action_box);
+		else
+			dsync_worker_delete_dir(action_worker, action_box);
+		break;
+	}
+}
+
+static bool
+dsync_mailbox_list_is_empty(const ARRAY_TYPE(dsync_mailbox) *boxes_arr)
+{
+	struct dsync_mailbox *const *boxes;
+	unsigned int count;
+
+	boxes = array_get(boxes_arr, &count);
+	if (count == 0)
+		return TRUE;
+	if (count == 1 && strcasecmp(boxes[0]->name, "INBOX") == 0 &&
+	    boxes[0]->message_count == 0)
+		return TRUE;
+	return FALSE;
+}
+
+static void dsync_brain_sync_mailboxes(struct dsync_brain *brain)
+{
+	struct dsync_mailbox *const *src_boxes, *const *dest_boxes;
+	struct dsync_mailbox *action_box = NULL;
+	struct dsync_worker *action_worker = NULL;
+	unsigned int src, dest, src_count, dest_count;
+	enum dsync_brain_mailbox_action action;
+	bool src_deleted, dest_deleted;
+	int ret;
+
+	if (brain->backup &&
+	    dsync_mailbox_list_is_empty(&brain->src_mailbox_list->mailboxes) &&
+	    !dsync_mailbox_list_is_empty(&brain->dest_mailbox_list->mailboxes)) {
+		i_fatal(DSYNC_WRONG_DIRECTION_ERROR_MSG);
+	}
+
+	/* create/delete missing mailboxes. the mailboxes are sorted by
+	   GUID, so we can do this quickly. */
+	src_boxes = array_get(&brain->src_mailbox_list->mailboxes, &src_count);
+	dest_boxes = array_get(&brain->dest_mailbox_list->mailboxes, &dest_count);
+	for (src = dest = 0; src < src_count && dest < dest_count; ) {
+		action = DSYNC_BRAIN_MAILBOX_ACTION_NONE;
+		src_deleted = (src_boxes[src]->flags &
+			       DSYNC_MAILBOX_FLAG_DELETED_MAILBOX) != 0;
+		dest_deleted = (dest_boxes[dest]->flags &
+				DSYNC_MAILBOX_FLAG_DELETED_MAILBOX) != 0;
+		ret = dsync_mailbox_guid_cmp(src_boxes[src],
+					     dest_boxes[dest]);
+		if (ret < 0) {
+			/* exists only in source */
+			if (!src_deleted) {
+				action = DSYNC_BRAIN_MAILBOX_ACTION_CREATE;
+				action_worker = brain->dest_worker;
+				action_box = src_boxes[src];
+			}
+			src++;
+		} else if (ret > 0) {
+			/* exists only in dest */
+			if (!dest_deleted) {
+				action = DSYNC_BRAIN_MAILBOX_ACTION_CREATE;
+				action_worker = brain->src_worker;
+				action_box = dest_boxes[dest];
+			}
+			dest++;
+		} else if (src_deleted) {
+			/* delete from dest too */
+			if (!dest_deleted) {
+				action = DSYNC_BRAIN_MAILBOX_ACTION_DELETE;
+				action_worker = brain->dest_worker;
+				action_box = dest_boxes[dest];
+			}
+			src++; dest++;
+		} else if (dest_deleted) {
+			/* delete from src too */
+			action = DSYNC_BRAIN_MAILBOX_ACTION_DELETE;
+			action_worker = brain->src_worker;
+			action_box = src_boxes[src];
+			src++; dest++;
+		} else {
+			src++; dest++;
+		}
+		dsync_brain_mailbox_action(brain, action,
+					   action_worker, action_box);
+	}
+	for (; src < src_count; src++) {
+		if ((src_boxes[src]->flags &
+		     DSYNC_MAILBOX_FLAG_DELETED_MAILBOX) != 0)
+			continue;
+
+		dsync_brain_mailbox_action(brain,
+			DSYNC_BRAIN_MAILBOX_ACTION_CREATE,
+			brain->dest_worker, src_boxes[src]);
+	}
+	for (; dest < dest_count; dest++) {
+		if ((dest_boxes[dest]->flags &
+		     DSYNC_MAILBOX_FLAG_DELETED_MAILBOX) != 0)
+			continue;
+
+		dsync_brain_mailbox_action(brain,
+			DSYNC_BRAIN_MAILBOX_ACTION_CREATE,
+			brain->src_worker, dest_boxes[dest]);
+	}
+}
+
+static void dsync_brain_sync_dirs(struct dsync_brain *brain)
+{
+	struct dsync_mailbox *const *src_boxes, *const *dest_boxes, *action_box;
+	unsigned int src, dest, src_count, dest_count;
+	enum dsync_brain_mailbox_action action;
+	struct dsync_worker *action_worker = NULL;
+	bool src_deleted, dest_deleted;
+	int ret;
+
+	/* create/delete missing directories. */
+	src_boxes = array_get(&brain->src_mailbox_list->dirs, &src_count);
+	dest_boxes = array_get(&brain->dest_mailbox_list->dirs, &dest_count);
+	for (src = dest = 0; src < src_count && dest < dest_count; ) {
+		action = DSYNC_BRAIN_MAILBOX_ACTION_NONE;
+		action_box = NULL;
+
+		src_deleted = (src_boxes[src]->flags &
+			       DSYNC_MAILBOX_FLAG_DELETED_DIR) != 0;
+		dest_deleted = (dest_boxes[dest]->flags &
+				DSYNC_MAILBOX_FLAG_DELETED_DIR) != 0;
+		ret = memcmp(src_boxes[src]->name_sha1.guid,
+			     dest_boxes[dest]->name_sha1.guid,
+			     sizeof(src_boxes[src]->name_sha1.guid));
+		if (ret < 0) {
+			/* exists only in source */
+			if (!src_deleted) {
+				action = DSYNC_BRAIN_MAILBOX_ACTION_CREATE;
+				action_worker = brain->dest_worker;
+				action_box = src_boxes[src];
+			}
+			src++;
+		} else if (ret > 0) {
+			/* exists only in dest */
+			if (!dest_deleted) {
+				action = DSYNC_BRAIN_MAILBOX_ACTION_CREATE;
+				action_worker = brain->src_worker;
+				action_box = dest_boxes[dest];
+			}
+			dest++;
+		} else if (src_deleted) {
+			/* delete from dest too */
+			if (!dest_deleted) {
+				action = DSYNC_BRAIN_MAILBOX_ACTION_DELETE;
+				action_worker = brain->dest_worker;
+				action_box = dest_boxes[dest];
+			}
+			src++; dest++;
+		} else if (dest_deleted) {
+			/* delete from src too */
+			action = DSYNC_BRAIN_MAILBOX_ACTION_DELETE;
+			action_worker = brain->src_worker;
+			action_box = src_boxes[src];
+			src++; dest++;
+		} else {
+			src++; dest++;
+		}
+		i_assert(action_box == NULL ||
+			 dsync_mailbox_is_noselect(action_box));
+		dsync_brain_mailbox_action(brain, action,
+					   action_worker, action_box);
+	}
+	for (; src < src_count; src++) {
+		if ((src_boxes[src]->flags &
+		     DSYNC_MAILBOX_FLAG_DELETED_DIR) != 0)
+			continue;
+
+		dsync_brain_mailbox_action(brain,
+			DSYNC_BRAIN_MAILBOX_ACTION_CREATE,
+			brain->dest_worker, src_boxes[src]);
+	}
+	for (; dest < dest_count; dest++) {
+		if ((dest_boxes[dest]->flags &
+		     DSYNC_MAILBOX_FLAG_DELETED_DIR) != 0)
+			continue;
+
+		dsync_brain_mailbox_action(brain,
+			DSYNC_BRAIN_MAILBOX_ACTION_CREATE,
+			brain->src_worker, dest_boxes[dest]);
+	}
+}
+
+static bool
+dsync_brain_is_unsubscribed(struct dsync_brain_subs_list *list,
+			    const struct dsync_worker_subscription *subs,
+			    time_t *last_change_r)
+{
+	const struct dsync_worker_unsubscription *unsubs;
+	struct dsync_worker_unsubscription lookup;
+
+	lookup.ns_prefix = subs->ns_prefix;
+	dsync_str_sha_to_guid(subs->storage_name, &lookup.name_sha1);
+	unsubs = array_bsearch(&list->unsubscriptions, &lookup,
+			       dsync_worker_unsubscription_cmp);
+	if (unsubs == NULL) {
+		*last_change_r = 0;
+		return FALSE;
+	} else if (unsubs->last_change <= subs->last_change) {
+		*last_change_r = subs->last_change;
+		return FALSE;
+	} else {
+		*last_change_r = unsubs->last_change;
+		return TRUE;
+	}
+}
+
+static void dsync_brain_sync_subscriptions(struct dsync_brain *brain)
+{
+	const struct dsync_worker_subscription *src_subs, *dest_subs;
+	const struct dsync_worker_subscription *action_subs;
+	struct dsync_worker *action_worker;
+	unsigned int src, dest, src_count, dest_count;
+	time_t last_change;
+	bool subscribe;
+	int ret;
+
+	/* subscriptions are sorted by name. */
+	src_subs = array_get(&brain->src_subs_list->subscriptions, &src_count);
+	dest_subs = array_get(&brain->dest_subs_list->subscriptions, &dest_count);
+	for (src = dest = 0;; ) {
+		if (src == src_count) {
+			if (dest == dest_count)
+				break;
+			ret = 1;
+		} else if (dest == dest_count) {
+			ret = -1;
+		} else {
+			ret = strcmp(src_subs[src].vname,
+				     dest_subs[dest].vname);
+			if (ret == 0) {
+				src++; dest++;
+				continue;
+			}
+		}
+
+		if (ret < 0) {
+			/* subscribed only in source */
+			action_subs = &src_subs[src];
+			if (dsync_brain_is_unsubscribed(brain->dest_subs_list,
+							&src_subs[src],
+							&last_change)) {
+				action_worker = brain->src_worker;
+				subscribe = FALSE;
+			} else {
+				action_worker = brain->dest_worker;
+				subscribe = TRUE;
+			}
+			src++;
+		} else {
+			/* subscribed only in dest */
+			action_subs = &dest_subs[dest];
+			if (dsync_brain_is_unsubscribed(brain->src_subs_list,
+							&dest_subs[dest],
+							&last_change)) {
+				action_worker = brain->dest_worker;
+				subscribe = FALSE;
+			} else {
+				action_worker = brain->src_worker;
+				subscribe = TRUE;
+			}
+			dest++;
+		}
+
+		if (brain->backup && action_worker == brain->src_worker) {
+			/* backup mode: switch action */
+			action_worker = brain->dest_worker;
+			subscribe = !subscribe;
+			last_change = ioloop_time;
+		}
+		dsync_worker_set_subscribed(action_worker, action_subs->vname,
+					    last_change, subscribe);
+	}
+}
+
+static bool dsync_mailbox_has_changed_msgs(struct dsync_brain *brain,
+					   const struct dsync_mailbox *box1,
+					   const struct dsync_mailbox *box2)
+{
+	const char *name = *box1->name != '\0' ? box1->name : box2->name;
+
+	if (box1->uid_validity != box2->uid_validity) {
+		if (brain->verbose) {
+			i_info("%s: uidvalidity changed: %u != %u", name,
+			       box1->uid_validity, box2->uid_validity);
+		}
+		return TRUE;
+	}
+	if (box1->uid_next != box2->uid_next) {
+		if (brain->verbose) {
+			i_info("%s: uidnext changed: %u != %u", name,
+			       box1->uid_next, box2->uid_next);
+		}
+		return TRUE;
+	}
+	if (box1->highest_modseq != box2->highest_modseq) {
+		if (brain->verbose) {
+			i_info("%s: highest_modseq changed: %llu != %llu", name,
+			       (unsigned long long)box1->highest_modseq,
+			       (unsigned long long)box2->highest_modseq);
+		}
+		return TRUE;
+	}
+	if (box1->message_count != box2->message_count) {
+		if (brain->verbose) {
+			i_info("%s: message_count changed: %u != %u", name,
+			       box1->message_count, box2->message_count);
+		}
+		return TRUE;
+	}
+	return FALSE;
+}
+
+static bool dsync_mailbox_has_changes(struct dsync_brain *brain,
+				      const struct dsync_mailbox *box1,
+				      const struct dsync_mailbox *box2)
+{
+	if (strcmp(box1->name, box2->name) != 0)
+		return TRUE;
+	return dsync_mailbox_has_changed_msgs(brain, box1, box2);
+}
+
+static void
+dsync_brain_get_changed_mailboxes(struct dsync_brain *brain,
+				  ARRAY_TYPE(dsync_brain_mailbox) *brain_boxes,
+				  bool full_sync)
+{
+	struct dsync_mailbox *const *src_boxes, *const *dest_boxes;
+	struct dsync_brain_mailbox *brain_box;
+	unsigned int src, dest, src_count, dest_count;
+	bool src_deleted, dest_deleted;
+	int ret;
+
+	src_boxes = array_get(&brain->src_mailbox_list->mailboxes, &src_count);
+	dest_boxes = array_get(&brain->dest_mailbox_list->mailboxes, &dest_count);
+
+	for (src = dest = 0; src < src_count && dest < dest_count; ) {
+		src_deleted = (src_boxes[src]->flags &
+			       DSYNC_MAILBOX_FLAG_DELETED_MAILBOX) != 0;
+		dest_deleted = (dest_boxes[dest]->flags &
+				DSYNC_MAILBOX_FLAG_DELETED_MAILBOX) != 0;
+
+		ret = dsync_mailbox_guid_cmp(src_boxes[src], dest_boxes[dest]);
+		if (ret == 0) {
+			if ((full_sync ||
+			     dsync_mailbox_has_changes(brain, src_boxes[src],
+						       dest_boxes[dest])) &&
+			    !src_deleted && !dest_deleted) {
+				brain_box = array_append_space(brain_boxes);
+				brain_box->box = *src_boxes[src];
+
+				brain_box->box.highest_modseq =
+					I_MAX(src_boxes[src]->highest_modseq,
+					      dest_boxes[dest]->highest_modseq);
+				brain_box->box.uid_next =
+					I_MAX(src_boxes[src]->uid_next,
+					      dest_boxes[dest]->uid_next);
+				brain_box->src = src_boxes[src];
+				brain_box->dest = dest_boxes[dest];
+			}
+			src++; dest++;
+		} else if (ret < 0) {
+			/* exists only in source */
+			if (!src_deleted) {
+				brain_box = array_append_space(brain_boxes);
+				brain_box->box = *src_boxes[src];
+				brain_box->src = src_boxes[src];
+				if (brain->verbose) {
+					i_info("%s: only in source (guid=%s)",
+					       brain_box->box.name,
+					       dsync_guid_to_str(&brain_box->box.mailbox_guid));
+				}
+			}
+ 			src++;
+		} else {
+			/* exists only in dest */
+			if (!dest_deleted) {
+				brain_box = array_append_space(brain_boxes);
+				brain_box->box = *dest_boxes[dest];
+				brain_box->dest = dest_boxes[dest];
+				if (brain->verbose) {
+					i_info("%s: only in dest (guid=%s)",
+					       brain_box->box.name,
+					       dsync_guid_to_str(&brain_box->box.mailbox_guid));
+				}
+			}
+			dest++;
+		}
+	}
+	for (; src < src_count; src++) {
+		if ((src_boxes[src]->flags &
+		     DSYNC_MAILBOX_FLAG_DELETED_MAILBOX) != 0)
+			continue;
+
+		brain_box = array_append_space(brain_boxes);
+		brain_box->box = *src_boxes[src];
+		brain_box->src = src_boxes[src];
+		if (brain->verbose) {
+			i_info("%s: only in source (guid=%s)",
+			       brain_box->box.name,
+			       dsync_guid_to_str(&brain_box->box.mailbox_guid));
+		}
+	}
+	for (; dest < dest_count; dest++) {
+		if ((dest_boxes[dest]->flags &
+		     DSYNC_MAILBOX_FLAG_DELETED_MAILBOX) != 0)
+			continue;
+
+		brain_box = array_append_space(brain_boxes);
+		brain_box->box = *dest_boxes[dest];
+		brain_box->dest = dest_boxes[dest];
+		if (brain->verbose) {
+			i_info("%s: only in dest (guid=%s)",
+			       brain_box->box.name,
+			       dsync_guid_to_str(&brain_box->box.mailbox_guid));
+		}
+	}
+}
+
+static bool dsync_brain_sync_msgs(struct dsync_brain *brain)
+{
+	ARRAY_TYPE(dsync_brain_mailbox) mailboxes;
+	pool_t pool;
+	bool ret;
+
+	pool = pool_alloconly_create("dsync changed mailboxes", 10240);
+	p_array_init(&mailboxes, pool, 128);
+	dsync_brain_get_changed_mailboxes(brain, &mailboxes,
+		(brain->flags & DSYNC_BRAIN_FLAG_FULL_SYNC) != 0);
+	if (array_count(&mailboxes) > 0) {
+		brain->mailbox_sync =
+			dsync_brain_msg_sync_init(brain, &mailboxes);
+		dsync_brain_msg_sync_more(brain->mailbox_sync);
+		ret = TRUE;
+	} else {
+		ret = FALSE;
+	}
+	pool_unref(&pool);
+	return ret;
+}
+
+static void
+dsync_brain_sync_rename_mailbox(struct dsync_brain *brain,
+				const struct dsync_brain_mailbox *mailbox)
+{
+	if (mailbox->src->last_change > mailbox->dest->last_change ||
+	    brain->backup) {
+		dsync_worker_rename_mailbox(brain->dest_worker,
+					    &mailbox->box.mailbox_guid,
+					    mailbox->src);
+	} else {
+		dsync_worker_rename_mailbox(brain->src_worker,
+					    &mailbox->box.mailbox_guid,
+					    mailbox->dest);
+	}
+}
+
+static void
+dsync_brain_sync_update_mailboxes(struct dsync_brain *brain)
+{
+	const struct dsync_brain_mailbox *mailbox;
+	bool failed_changes = dsync_brain_has_unexpected_changes(brain) ||
+		dsync_worker_has_failed(brain->src_worker) ||
+		dsync_worker_has_failed(brain->dest_worker);
+
+	if (brain->mailbox_sync == NULL) {
+		/* no mailboxes changed */
+		return;
+	}
+
+	array_foreach(&brain->mailbox_sync->mailboxes, mailbox) {
+		/* don't update mailboxes if any changes had failed.
+		   for example if some messages couldn't be saved, we don't
+		   want to increase the next_uid to jump over them */
+		if (!brain->backup && !failed_changes) {
+			dsync_worker_update_mailbox(brain->src_worker,
+						    &mailbox->box);
+		}
+		if (!failed_changes) {
+			dsync_worker_update_mailbox(brain->dest_worker,
+						    &mailbox->box);
+		}
+
+		if (mailbox->src != NULL && mailbox->dest != NULL &&
+		    strcmp(mailbox->src->name, mailbox->dest->name) != 0)
+			dsync_brain_sync_rename_mailbox(brain, mailbox);
+	}
+}
+
+static void dsync_brain_worker_finished(bool success, void *context)
+{
+	struct dsync_brain *brain = context;
+
+	switch (brain->state) {
+	case DSYNC_STATE_SYNC_MSGS_FLUSH:
+	case DSYNC_STATE_SYNC_MSGS_FLUSH2:
+	case DSYNC_STATE_SYNC_FLUSH:
+	case DSYNC_STATE_SYNC_FLUSH2:
+		break;
+	default:
+		i_panic("dsync brain state=%d", brain->state);
+	}
+
+	if (!success)
+		dsync_brain_fail(brain);
+
+	brain->state++;
+	if (brain->to == NULL && (brain->flags & DSYNC_BRAIN_FLAG_LOCAL) == 0)
+		brain->to = timeout_add(0, dsync_brain_sync, brain);
+}
+
+void dsync_brain_sync(struct dsync_brain *brain)
+{
+	if (dsync_worker_has_failed(brain->src_worker) ||
+	    dsync_worker_has_failed(brain->dest_worker)) {
+		/* we can't safely continue, especially with backup */
+		return;
+	}
+
+	if (brain->to != NULL)
+		timeout_remove(&brain->to);
+	switch (brain->state) {
+	case DSYNC_STATE_GET_MAILBOXES:
+		i_assert(brain->src_mailbox_list == NULL);
+		brain->src_mailbox_list =
+			dsync_brain_mailbox_list_init(brain, brain->src_worker);
+		brain->dest_mailbox_list =
+			dsync_brain_mailbox_list_init(brain, brain->dest_worker);
+		dsync_worker_mailbox_input(brain->src_mailbox_list);
+		dsync_worker_mailbox_input(brain->dest_mailbox_list);
+		break;
+	case DSYNC_STATE_GET_SUBSCRIPTIONS:
+		i_assert(brain->src_subs_list == NULL);
+		brain->src_subs_list =
+			dsync_brain_subs_list_init(brain, brain->src_worker);
+		brain->dest_subs_list =
+			dsync_brain_subs_list_init(brain, brain->dest_worker);
+		dsync_worker_subs_input(brain->src_subs_list);
+		dsync_worker_subs_input(brain->dest_subs_list);
+		break;
+	case DSYNC_STATE_SYNC_MAILBOXES:
+		dsync_worker_set_input_callback(brain->src_worker, NULL, NULL);
+		dsync_worker_set_input_callback(brain->dest_worker, NULL, NULL);
+
+		dsync_brain_sync_mailboxes(brain);
+		dsync_brain_sync_dirs(brain);
+		brain->state++;
+		/* fall through */
+	case DSYNC_STATE_SYNC_SUBSCRIPTIONS:
+		dsync_brain_sync_subscriptions(brain);
+		brain->state++;
+		/* fall through */
+	case DSYNC_STATE_SYNC_MSGS:
+		if (dsync_brain_sync_msgs(brain))
+			break;
+		brain->state++;
+		/* no mailboxes changed */
+	case DSYNC_STATE_SYNC_MSGS_FLUSH:
+		/* wait until all saves are done, so we don't try to close
+		   the mailbox too early */
+		dsync_worker_finish(brain->src_worker,
+				    dsync_brain_worker_finished, brain);
+		dsync_worker_finish(brain->dest_worker,
+				    dsync_brain_worker_finished, brain);
+		break;
+	case DSYNC_STATE_SYNC_MSGS_FLUSH2:
+		break;
+	case DSYNC_STATE_SYNC_UPDATE_MAILBOXES:
+		dsync_brain_sync_update_mailboxes(brain);
+		brain->state++;
+		/* fall through */
+	case DSYNC_STATE_SYNC_FLUSH:
+		dsync_worker_finish(brain->src_worker,
+				    dsync_brain_worker_finished, brain);
+		dsync_worker_finish(brain->dest_worker,
+				    dsync_brain_worker_finished, brain);
+		break;
+	case DSYNC_STATE_SYNC_FLUSH2:
+		break;
+	case DSYNC_STATE_SYNC_END:
+		io_loop_stop(current_ioloop);
+		break;
+	default:
+		i_unreached();
+	}
+}
+
+void dsync_brain_sync_all(struct dsync_brain *brain)
+{
+	enum dsync_state old_state;
+
+	while (brain->state != DSYNC_STATE_SYNC_END) {
+		old_state = brain->state;
+		dsync_brain_sync(brain);
+
+		if (dsync_worker_has_failed(brain->src_worker) ||
+		    dsync_worker_has_failed(brain->dest_worker))
+			break;
+
+		i_assert(brain->state != old_state);
+	}
+}
+
+bool dsync_brain_has_unexpected_changes(struct dsync_brain *brain)
+{
+	return brain->unexpected_changes ||
+		dsync_worker_has_unexpected_changes(brain->src_worker) ||
+		dsync_worker_has_unexpected_changes(brain->dest_worker);
+}
+
+bool dsync_brain_has_failed(struct dsync_brain *brain)
+{
+	return brain->failed ||
+		dsync_worker_has_failed(brain->src_worker) ||
+		dsync_worker_has_failed(brain->dest_worker);
+}

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/doveadm/dsync/dsync-brain.h	Thu Dec 29 14:43:45 2011 +0200
@@ -0,0 +1,28 @@
+#ifndef DSYNC_BRAIN_H
+#define DSYNC_BRAIN_H
+
+enum dsync_brain_flags {
+	DSYNC_BRAIN_FLAG_FULL_SYNC	= 0x01,
+	DSYNC_BRAIN_FLAG_VERBOSE	= 0x02,
+	/* Run in backup mode. All changes from src are forced into dest,
+	   discarding any potential changes in dest. */
+	DSYNC_BRAIN_FLAG_BACKUP		= 0x04,
+	/* Run in "local mode". Don't use ioloop. */
+	DSYNC_BRAIN_FLAG_LOCAL		= 0x08
+};
+
+struct dsync_worker;
+
+struct dsync_brain *
+dsync_brain_init(struct dsync_worker *src_worker,
+		 struct dsync_worker *dest_worker,
+		 const char *mailbox, enum dsync_brain_flags flags);
+int dsync_brain_deinit(struct dsync_brain **brain);
+
+void dsync_brain_sync(struct dsync_brain *brain);
+void dsync_brain_sync_all(struct dsync_brain *brain);
+
+bool dsync_brain_has_unexpected_changes(struct dsync_brain *brain);
+bool dsync_brain_has_failed(struct dsync_brain *brain);
+
+#endif

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/doveadm/dsync/dsync-data.c	Thu Dec 29 14:43:45 2011 +0200
@@ -0,0 +1,146 @@
+/* Copyright (c) 2009-2011 Dovecot authors, see the included COPYING file */
+
+#include "lib.h"
+#include "array.h"
+#include "hex-binary.h"
+#include "sha1.h"
+#include "dsync-data.h"
+
+struct dsync_mailbox *
+dsync_mailbox_dup(pool_t pool, const struct dsync_mailbox *box)
+{
+	struct dsync_mailbox *dest;
+	const struct mailbox_cache_field *cache_fields = NULL;
+	struct mailbox_cache_field *dup;
+	unsigned int i, count = 0;
+
+	dest = p_new(pool, struct dsync_mailbox, 1);
+	*dest = *box;
+	dest->name = p_strdup(pool, box->name);
+
+	if (array_is_created(&box->cache_fields))
+		cache_fields = array_get(&box->cache_fields, &count);
+	if (count == 0)
+		memset(&dest->cache_fields, 0, sizeof(dest->cache_fields));
+	else {
+		p_array_init(&dest->cache_fields, pool, count);
+		for (i = 0; i < count; i++) {
+			dup = array_append_space(&dest->cache_fields);
+			*dup = cache_fields[i];
+			dup->name = p_strdup(pool, dup->name);
+		}
+	}
+	return dest;
+}
+
+struct dsync_message *
+dsync_message_dup(pool_t pool, const struct dsync_message *msg)
+{
+	struct dsync_message *dest;
+	const char **keywords;
+	unsigned int i, count;
+
+	dest = p_new(pool, struct dsync_message, 1);
+	*dest = *msg;
+	dest->guid = p_strdup(pool, msg->guid);
+	if (msg->keywords != NULL) {
+		count = str_array_length(msg->keywords);
+		keywords = p_new(pool, const char *, count+1);
+		for (i = 0; i < count; i++)
+			keywords[i] = p_strdup(pool, msg->keywords[i]);
+		dest->keywords = keywords;
+	}
+	return dest;
+}
+
+int dsync_mailbox_guid_cmp(const struct dsync_mailbox *box1,
+			   const struct dsync_mailbox *box2)
+{
+	return memcmp(box1->mailbox_guid.guid, box2->mailbox_guid.guid,
+		      sizeof(box1->mailbox_guid.guid));
+}
+
+int dsync_mailbox_p_guid_cmp(struct dsync_mailbox *const *box1,
+			     struct dsync_mailbox *const *box2)
+{
+	return dsync_mailbox_guid_cmp(*box1, *box2);
+}
+
+int dsync_mailbox_name_sha1_cmp(const struct dsync_mailbox *box1,
+				const struct dsync_mailbox *box2)
+{
+	int ret;
+
+	ret = memcmp(box1->name_sha1.guid, box2->name_sha1.guid,
+		     sizeof(box1->name_sha1.guid));
+	if (ret != 0)
+		return ret;
+
+	return strcmp(box1->name, box2->name);
+}
+
+int dsync_mailbox_p_name_sha1_cmp(struct dsync_mailbox *const *box1,
+				  struct dsync_mailbox *const *box2)
+{
+	return dsync_mailbox_name_sha1_cmp(*box1, *box2);
+}
+
+bool dsync_keyword_list_equals(const char *const *k1, const char *const *k2)
+{
+	unsigned int i;
+
+	if (k1 == NULL)
+		return k2 == NULL || k2[0] == NULL;
+	if (k2 == NULL)
+		return k1[0] == NULL;
+
+	for (i = 0;; i++) {
+		if (k1[i] == NULL)
+			return k2[i] == NULL;
+		if (k2[i] == NULL)
+			return FALSE;
+
+		if (strcasecmp(k1[i], k2[i]) != 0)
+			return FALSE;
+	}
+}
+
+bool dsync_guid_equals(const mailbox_guid_t *guid1,
+		       const mailbox_guid_t *guid2)
+{
+	return memcmp(guid1->guid, guid2->guid, sizeof(guid1->guid)) == 0;
+}
+
+int dsync_guid_cmp(const mailbox_guid_t *guid1, const mailbox_guid_t *guid2)
+{
+	return memcmp(guid1->guid, guid2->guid, sizeof(guid1->guid));
+}
+
+const char *dsync_guid_to_str(const mailbox_guid_t *guid)
+{
+	return guid_128_to_string(guid->guid);
+}
+
+const char *dsync_get_guid_128_str(const char *guid, unsigned char *dest,
+				   unsigned int dest_len)
+{
+	guid_128_t guid_128;
+	buffer_t guid_128_buf;
+
+	i_assert(dest_len >= GUID_128_SIZE * 2 + 1);
+	buffer_create_data(&guid_128_buf, dest, dest_len);
+	mail_generate_guid_128_hash(guid, guid_128);
+	if (guid_128_is_empty(guid_128))
+		return "";
+	binary_to_hex_append(&guid_128_buf, guid_128, sizeof(guid_128));
+	buffer_append_c(&guid_128_buf, '\0');
+	return guid_128_buf.data;
+}
+
+void dsync_str_sha_to_guid(const char *str, mailbox_guid_t *guid)
+{
+	unsigned char sha[SHA1_RESULTLEN];
+
+	sha1_get_digest(str, strlen(str), sha);
+	memcpy(guid->guid, sha, I_MIN(sizeof(guid->guid), sizeof(sha)));
+}

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/doveadm/dsync/dsync-data.h	Thu Dec 29 14:43:45 2011 +0200
@@ -0,0 +1,84 @@
+#ifndef DSYNC_DATA_H
+#define DSYNC_DATA_H
+
+#include "mail-storage.h"
+
+typedef struct {
+	guid_128_t guid;
+} mailbox_guid_t;
+ARRAY_DEFINE_TYPE(mailbox_guid, mailbox_guid_t);
+
+enum dsync_mailbox_flags {
+	DSYNC_MAILBOX_FLAG_NOSELECT		= 0x01,
+	DSYNC_MAILBOX_FLAG_DELETED_MAILBOX	= 0x02,
+	DSYNC_MAILBOX_FLAG_DELETED_DIR		= 0x04
+};
+
+struct dsync_mailbox {
+	const char *name;
+	char name_sep;
+	/* 128bit SHA1 sum of mailbox name */
+	mailbox_guid_t name_sha1;
+	/* Mailbox's GUID. Full of zero with \Noselect mailboxes. */
+	mailbox_guid_t mailbox_guid;
+
+	uint32_t uid_validity, uid_next, message_count, first_recent_uid;
+	uint64_t highest_modseq;
+	/* if mailbox is deleted, this is the deletion timestamp.
+	   otherwise it's the last rename timestamp. */
+	time_t last_change;
+	enum dsync_mailbox_flags flags;
+	ARRAY_TYPE(mailbox_cache_field) cache_fields;
+};
+ARRAY_DEFINE_TYPE(dsync_mailbox, struct dsync_mailbox *);
+#define dsync_mailbox_is_noselect(dsync_box) \
+	(((dsync_box)->flags & DSYNC_MAILBOX_FLAG_NOSELECT) != 0)
+
+/* dsync_worker_msg_iter_next() returns also all expunged messages from
+   the end of mailbox with this flag set. The GUIDs are 128 bit GUIDs saved
+   to transaction log (mail_generate_guid_128_hash()). */
+#define DSYNC_MAIL_FLAG_EXPUNGED 0x10000000
+
+struct dsync_message {
+	const char *guid;
+	uint32_t uid;
+	enum mail_flags flags;
+	/* keywords are sorted by name */
+	const char *const *keywords;
+	uint64_t modseq;
+	time_t save_date;
+};
+
+struct dsync_msg_static_data {
+	const char *pop3_uidl;
+	time_t received_date;
+	struct istream *input;
+};
+
+struct dsync_mailbox *
+dsync_mailbox_dup(pool_t pool, const struct dsync_mailbox *box);
+
+struct dsync_message *
+dsync_message_dup(pool_t pool, const struct dsync_message *msg);
+
+int dsync_mailbox_guid_cmp(const struct dsync_mailbox *box1,
+			   const struct dsync_mailbox *box2);
+int dsync_mailbox_p_guid_cmp(struct dsync_mailbox *const *box1,
+			     struct dsync_mailbox *const *box2);
+
+int dsync_mailbox_name_sha1_cmp(const struct dsync_mailbox *box1,
+				const struct dsync_mailbox *box2);
+int dsync_mailbox_p_name_sha1_cmp(struct dsync_mailbox *const *box1,
+				  struct dsync_mailbox *const *box2);
+
+bool dsync_keyword_list_equals(const char *const *k1, const char *const *k2);
+
+bool dsync_guid_equals(const mailbox_guid_t *guid1,
+		       const mailbox_guid_t *guid2);
+int dsync_guid_cmp(const mailbox_guid_t *guid1, const mailbox_guid_t *guid2);
+const char *dsync_guid_to_str(const mailbox_guid_t *guid);
+const char *dsync_get_guid_128_str(const char *guid, unsigned char *dest,
+				   unsigned int dest_len);
+void dsync_str_sha_to_guid(const char *str, mailbox_guid_t *guid);
+
+#endif

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/src/doveadm/dsync/dsync-proxy-client.c	Thu Dec 29 14:43:45 2011 +0200
@@ -0,0 +1,1191 @@
+/* Copyright (c) 2009-2011 Dovecot authors, see the included COPYING file */
+
+#include "lib.h"
+#include "array.h"
+#include "aqueue.h"
+#include "fd-set-nonblock.h"
+#include "istream.h"
+#include "istream-dot.h"
+#include "ostream.h"
+#include "str.h"
+#include "strescape.h"
+#include "imap-util.h"
+#include "dsync-proxy.h"
+#include "dsync-worker-private.h"
+
+#include <stdlib.h>
+#include <unistd.h>
+
+#define OUTBUF_THROTTLE_SIZE (1024*64)
+
+enum proxy_client_request_type {
+	PROXY_CLIENT_REQUEST_TYPE_COPY,
+	PROXY_CLIENT_REQUEST_TYPE_GET,
+	PROXY_CLIENT_REQUEST_TYPE_FINISH
+};
+
+struct proxy_client_request {
+	enum proxy_client_request_type type;
+	uint32_t uid;
+	union {
+		dsync_worker_msg_callback_t *get;
+		dsync_worker_copy_callback_t *copy;
+		dsync_worker_finish_callback_t *finish;
+	} callback;
+	void *context;
+};
+
+struct proxy_client_dsync_worker_mailbox_iter {
+	struct dsync_worker_mailbox_iter iter;
+	pool_t pool;
+};
+
+struct proxy_client_dsync_worker_subs_iter {
+	struct dsync_worker_subs_iter iter;
+	pool_t pool;
+};
+
+struct proxy_client_dsync_worker {
+	struct dsync_worker worker;
+	int fd_in, fd_out;
+	struct io *io;
+	struct istream *input;
+	struct ostream *output;
+	struct timeout *to, *to_input;
+
+	mailbox_guid_t selected_box_guid;
+
+	dsync_worker_save_callback_t *save_callback;
+	void *save_context;
+	struct istream *save_input;
+	struct io *save_io;
+	bool save_input_last_lf;
+
+	pool_t msg_get_pool;
+	struct dsync_msg_static_data msg_get_data;
+	ARRAY_DEFINE(request_array, struct proxy_client_request);
+	struct aqueue *request_queue;
+	string_t *pending_commands;
+
+	unsigned int handshake_received:1;
+	unsigned int finishing:1;
+	unsigned int finished:1;
+};
+
+extern struct dsync_worker_vfuncs proxy_client_dsync_worker;
+
+static void proxy_client_worker_input(struct proxy_client_dsync_worker *worker);
+static void proxy_client_send_stream(struct proxy_client_dsync_worker *worker);
+
+static void proxy_client_fail(struct proxy_client_dsync_worker *worker)
+{
+	i_stream_close(worker->input);
+	dsync_worker_set_failure(&worker->worker);
+	io_loop_stop(current_ioloop);
+}
+
+static int
+proxy_client_worker_read_line(struct proxy_client_dsync_worker *worker,
+			      const char **line_r)
+{
+	if (worker->worker.failed)
+		return -1;
+
+	*line_r = i_stream_read_next_line(worker->input);
+	if (*line_r == NULL) {
+		if (worker->input->stream_errno != 0) {
+			errno = worker->input->stream_errno;
+			i_error("read() from worker server failed: %m");
+			dsync_worker_set_failure(&worker->worker);
+			return -1;
+		}
+		if (worker->input->eof) {
+			if (!worker->finished)
+				i_error("read() from worker server failed: EOF");
+			dsync_worker_set_failure(&worker->worker);
+			return -1;
+		}
+	}
+	if (*line_r == NULL)
+		return 0;
+
+	if (!worker->handshake_received) {
+		if (strcmp(*line_r, DSYNC_PROXY_SERVER_GREETING_LINE) != 0) {
+			i_error("Invalid server handshake: %s", *line_r);
+			dsync_worker_set_failure(&worker->worker);
+			return -1;
+		}
+		worker->handshake_received = TRUE;
+		return proxy_client_worker_read_line(worker, line_r);
+	}
+	return 1;
+}
+
+static void
+proxy_client_worker_msg_get_finish(struct proxy_client_dsync_worker *worker)
+{
+	worker->msg_get_data.input = NULL;
+	worker->io = io_add(worker->fd_in, IO_READ,
+			    proxy_client_worker_input, worker);
+
+	/* some input may already be buffered. note that we may be coming here
+	   from the input function itself, in which case this timeout must not
+	   be called (we'll remove it later) */
+	if (worker->to_input == NULL) {
+		worker->to_input =
+			timeout_add(0, proxy_client_worker_input, worker);
+	}
+}
+
+static void
+proxy_client_worker_read_to_eof(struct proxy_client_dsync_worker *worker)
+{
+	struct istream *input = worker->msg_get_data.input;
+	const unsigned char *data;
+	size_t size;
+	int ret;
+
+	while ((ret = i_stream_read_data(input, &data, &size, 0)) > 0)
+		i_stream_skip(input, size);
+	if (ret == -1) {
+		i_stream_unref(&input);
+		io_remove(&worker->io);
+		proxy_client_worker_msg_get_finish(worker);
+	}
+	timeout_reset(worker->to);
+}
+
+static void
+proxy_client_worker_msg_get_done(struct proxy_client_dsync_worker *worker)
+{
+	struct istream *input = worker->msg_get_data.input;
+
+	i_assert(worker->io == NULL);
+
+	if (input->eof)
+		proxy_client_worker_msg_get_finish(worker);
+	else {
+		/* saving read the message only partially. we'll need to read
+		   the input until EOF or we'll start treating the input as
+		   commands. */
+		worker->io = io_add(worker->fd_in, IO_READ,
+				    proxy_client_worker_read_to_eof, worker);
+		worker->msg_get_data.input =
+			i_stream_create_dot(worker->input, FALSE);
+	}
+}
+
+static bool
+proxy_client_worker_next_copy(struct proxy_client_dsync_worker *worker,
+			      const struct proxy_client_request *request,
+			      const char *line)
+{
+	uint32_t uid;
+	bool success;
+
+	if (line[0] == '1' && line[1] == '\t')
+		success = TRUE;
+	else if (line[0] == '0' && line[1] == '\t')
+		success = FALSE;
+	else {
+		i_error("msg-copy returned invalid input: %s", line);
+		proxy_client_fail(worker);
+		return FALSE;
+	}
+	uid = strtoul(line + 2, NULL, 10);
+	if (uid != request->uid) {
+		i_error("msg-copy returned invalid uid: %u != %u",
+			uid, request->uid);
+		proxy_client_fail(worker);
+		return FALSE;
+	}
+
+	request->callback.copy(success, request->context);
+	return TRUE;
+}
+
+static bool
+proxy_client_worker_next_msg_get(struct proxy_client_dsync_worker *worker,
+				 const struct proxy_client_request *request,
+				 const char *line)
+{
+	enum dsync_msg_get_result result = DSYNC_MSG_GET_RESULT_FAILED;
+	const char *p, *error;
+	uint32_t uid;
+
+	p_clear(worker->msg_get_pool);
+	switch (line[0]) {
+	case '1':
+		/* ok */
+		if (line[1] != '\t')
+			break;
+		line += 2;
+
+		if ((p = strchr(line, '\t')) == NULL)
+			break;
+		uid = strtoul(t_strcut(line, '\t'), NULL, 10);
+		line = p + 1;
+
+		if (uid != request->uid) {
+			i_error("msg-get returned invalid uid: %u != %u",
+				uid, request->uid);
+			proxy_client_fail(worker);
+			return FALSE;
+		}
+
+		if (dsync_proxy_msg_static_import(worker->msg_get_pool,
+						  line, &worker->msg_get_data,
+						  &error) < 0) {
+			i_error("Invalid msg-get static input: %s", error);
+			proxy_client_fail(worker);
+			return FALSE;
+		}
+		worker->msg_get_data.input =
+			i_stream_create_dot(worker->input, FALSE);
+		i_stream_set_destroy_callback(worker->msg_get_data.input,
+					      proxy_client_worker_msg_get_done,
+					      worker);
+		io_remove(&worker->io);
+		result = DSYNC_MSG_GET_RESULT_SUCCESS;
+		break;
+	case '0':
+		/* expunged */
+		result = DSYNC_MSG_GET_RESULT_EXPUNGED;
+		break;
+	default:
+		/* failure */
+		break;
+	}
+
+	request->callback.get(result, &worker->msg_get_data, request->context);
+	return worker->io != NULL && worker->msg_get_data.input == NULL;
+}
+
+static void
+proxy_client_worker_next_finish(struct proxy_client_dsync_worker *worker,
+				const struct proxy_client_request *request,
+				const char *line)
+{
+	bool success = TRUE;
+
+	i_assert(worker->finishing);
+	i_assert(!worker->finished);
+
+	worker->finishing = FALSE;
+	worker->finished = TRUE;
+
+	if (strcmp(line, "changes") == 0)
+		worker->worker.unexpected_changes = TRUE;
+	else if (strcmp(line, "fail") == 0)
+		success = FALSE;
+	else if (strcmp(line, "ok") != 0) {
+		i_error("Unexpected finish reply: %s", line);
+		success = FALSE;
+	}
+		
+	request->callback.finish(success, request->context);
+}
+
+static bool
+proxy_client_worker_next_reply(struct proxy_client_dsync_worker *worker,
+			       const char *line)
+{
+	const struct proxy_client_request *requests;
+	struct proxy_client_request request;
+	bool ret = TRUE;
+
+	i_assert(worker->msg_get_data.input == NULL);
+
+	if (aqueue_count(worker->request_queue) == 0) {
+		i_error("Unexpected reply from server: %s", line);
+		proxy_client_fail(worker);
+		return FALSE;
+	}
+
+	requests = array_idx(&worker->request_array, 0);
+	request = requests[aqueue_idx(worker->request_queue, 0)];
+	aqueue_delete_tail(worker->request_queue);
+
+	switch (request.type) {
+	case PROXY_CLIENT_REQUEST_TYPE_COPY:
+		ret = proxy_client_worker_next_copy(worker, &request, line);
+		break;
+	case PROXY_CLIENT_REQUEST_TYPE_GET:
+		ret = proxy_client_worker_next_msg_get(worker, &request, line);
+		break;
+	case PROXY_CLIENT_REQUEST_TYPE_FINISH:
+		proxy_client_worker_next_finish(worker, &request, line);
+		break;
+	}
+	return ret;
+}
+
+static void proxy_client_worker_input(struct proxy_client_dsync_worker *worker)
+{
+	const char *line;
+	int ret;
+
+	if (worker->to_input != NULL)
+		timeout_remove(&worker->to_input);
+
+	if (worker->worker.input_callback != NULL) {
+		worker->worker.input_callback(worker->worker.input_context);
+		timeout_reset(worker->to);
+		return;
+	}
+
+	while ((ret = proxy_client_worker_read_line(worker, &line)) > 0) {
+		if (!proxy_client_worker_next_reply(worker, line))
+			break;
+	}
+	if (ret < 0) {
+		/* try to continue */
+		proxy_client_worker_next_reply(worker, "");
+	}
+
+	if (worker->to_input != NULL) {
+		/* input stream's destroy callback was already called.
+		   don't get back here. */
+		timeout_remove(&worker->to_input);
+	}
+	timeout_reset(worker->to);
+}
+
+static int
+proxy_client_worker_output_real(struct proxy_client_dsync_worker *worker)
+{
+	int ret;
+
+	if ((ret = o_stream_flush(worker->output)) < 0)
+		return 1;
+
+	if (worker->save_input != NULL) {
+		/* proxy_client_worker_msg_save() hasn't finished yet. */
+		o_stream_cork(worker->output);
+		proxy_client_send_stream(worker);
+		if (worker->save_input != NULL) {
+			/* still unfinished, make sure we get called again */
+			return 0;
+		}
+	}
+
+	if (worker->worker.output_callback != NULL)
+		worker->worker.output_callback(worker->worker.output_context);
+	return ret;
+}
+
+static int proxy_client_worker_output(struct proxy_client_dsync_worker *worker)
+{
+	int ret;
+
+	ret = proxy_client_worker_output_real(worker);
+	timeout_reset(worker->to);
+	return ret;
+}
+
+static void
+proxy_client_worker_timeout(struct proxy_client_dsync_worker *worker)
+{
+	const char *reason;
+
+	if (worker->save_io != NULL)
+		reason = " (waiting for more input from mail being saved)";
+	else if (worker->save_input != NULL) {
+		size_t bytes = o_stream_get_buffer_used_size(worker->output);
+
+		reason = t_strdup_printf(" (waiting for output stream to flush, "
+					 "%"PRIuSIZE_T" bytes left)", bytes);
+	} else if (worker->msg_get_data.input != NULL) {
+		reason = " (waiting for MSG-GET message from remote)";
+	} else {
+		reason = "";
+	}
+	i_error("proxy client timed out%s", reason);
+	proxy_client_fail(worker);
+}
+
+struct dsync_worker *dsync_worker_init_proxy_client(int fd_in, int fd_out)
+{
+	struct proxy_client_dsync_worker *worker;
+
+	worker = i_new(struct proxy_client_dsync_worker, 1);
+	worker->worker.v = proxy_client_dsync_worker;
+	worker->fd_in = fd_in;
+	worker->fd_out = fd_out;
+	worker->to = timeout_add(DSYNC_PROXY_CLIENT_TIMEOUT_MSECS,
+				 proxy_client_worker_timeout, worker);
+	worker->io = io_add(fd_in, IO_READ, proxy_client_worker_input, worker);
+	worker->input = i_stream_create_fd(fd_in, (size_t)-1, FALSE);
+	worker->output = o_stream_create_fd(fd_out, (size_t)-1, FALSE);
+	o_stream_send_str(worker->output, DSYNC_PROXY_CLIENT_GREETING_LINE"\n");
+	/* we'll keep the output corked until flush is needed */
+	o_stream_cork(worker->output);
+	o_stream_set_flush_callback(worker->output, proxy_client_worker_output,
+				    worker);
+	fd_set_nonblock(fd_in, TRUE);
+	fd_set_nonblock(fd_out, TRUE);
+
+	worker->pending_commands = str_new(default_pool, 1024);
+	worker->msg_get_pool = pool_alloconly_create("dsync proxy msg", 128);
+	i_array_init(&worker->request_array, 64);
+	worker->request_queue = aqueue_init(&worker->request_array.arr);
+
+	return &worker->worker;
+}
+
+static void proxy_client_worker_deinit(struct dsync_worker *_worker)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+
+	timeout_remove(&worker->to);
+	if (worker->to_input != NULL)
+		timeout_remove(&worker->to_input);
+	if (worker->io != NULL)
+		io_remove(&worker->io);
+	i_stream_destroy(&worker->input);
+	o_stream_destroy(&worker->output);
+	if (close(worker->fd_in) < 0)
+		i_error("close(worker input) failed: %m");
+	if (worker->fd_in != worker->fd_out) {
+		if (close(worker->fd_out) < 0)
+			i_error("close(worker output) failed: %m");
+	}
+	aqueue_deinit(&worker->request_queue);
+	array_free(&worker->request_array);
+	pool_unref(&worker->msg_get_pool);
+	str_free(&worker->pending_commands);
+	i_free(worker);
+}
+
+static bool
+worker_is_output_stream_full(struct proxy_client_dsync_worker *worker)
+{
+	return o_stream_get_buffer_used_size(worker->output) >=
+		OUTBUF_THROTTLE_SIZE;
+}
+
+static bool proxy_client_worker_is_output_full(struct dsync_worker *_worker)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+
+	if (worker->save_input != NULL) {
+		/* we haven't finished sending a message save, so we're full. */
+		return TRUE;
+	}
+	return worker_is_output_stream_full(worker);
+}
+
+static int proxy_client_worker_output_flush(struct dsync_worker *_worker)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+	int ret = 1;
+
+	if (o_stream_flush(worker->output) < 0)
+		return -1;
+
+	o_stream_uncork(worker->output);
+	if (o_stream_get_buffer_used_size(worker->output) > 0)
+		return 0;
+
+	if (o_stream_send(worker->output, str_data(worker->pending_commands),
+			  str_len(worker->pending_commands)) < 0)
+		ret = -1;
+	str_truncate(worker->pending_commands, 0);
+	o_stream_cork(worker->output);
+	return ret;
+}
+
+static struct dsync_worker_mailbox_iter *
+proxy_client_worker_mailbox_iter_init(struct dsync_worker *_worker)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+	struct proxy_client_dsync_worker_mailbox_iter *iter;
+
+	iter = i_new(struct proxy_client_dsync_worker_mailbox_iter, 1);
+	iter->iter.worker = _worker;
+	iter->pool = pool_alloconly_create("proxy mailbox iter", 1024);
+	o_stream_send_str(worker->output, "BOX-LIST\n");
+	(void)proxy_client_worker_output_flush(_worker);
+	return &iter->iter;
+}
+
+static int
+proxy_client_worker_mailbox_iter_next(struct dsync_worker_mailbox_iter *_iter,
+				      struct dsync_mailbox *dsync_box_r)
+{
+	struct proxy_client_dsync_worker_mailbox_iter *iter =
+		(struct proxy_client_dsync_worker_mailbox_iter *)_iter;
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_iter->worker;
+	const char *line, *error;
+	int ret;
+
+	if ((ret = proxy_client_worker_read_line(worker, &line)) <= 0) {
+		if (ret < 0)
+			_iter->failed = TRUE;
+		return ret;
+	}
+
+	if ((line[0] == '+' || line[0] == '-') && line[1] == '\0') {
+		/* end of mailboxes */
+		if (line[0] == '-') {
+			i_error("Worker server's mailbox iteration failed");
+			_iter->failed = TRUE;
+		}
+		return -1;
+	}
+
+	p_clear(iter->pool);
+	if (dsync_proxy_mailbox_import(iter->pool, line,
+				       dsync_box_r, &error) < 0) {
+		i_error("Invalid mailbox input from worker server: %s", error);
+		_iter->failed = TRUE;
+		return -1;
+	}
+	return 1;
+}
+
+static int
+proxy_client_worker_mailbox_iter_deinit(struct dsync_worker_mailbox_iter *_iter)
+{
+	struct proxy_client_dsync_worker_mailbox_iter *iter =
+		(struct proxy_client_dsync_worker_mailbox_iter *)_iter;
+	int ret = _iter->failed ? -1 : 0;
+
+	pool_unref(&iter->pool);
+	i_free(iter);
+	return ret;
+}
+
+static struct dsync_worker_subs_iter *
+proxy_client_worker_subs_iter_init(struct dsync_worker *_worker)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+	struct proxy_client_dsync_worker_subs_iter *iter;
+
+	iter = i_new(struct proxy_client_dsync_worker_subs_iter, 1);
+	iter->iter.worker = _worker;
+	iter->pool = pool_alloconly_create("proxy subscription iter", 1024);
+	o_stream_send_str(worker->output, "SUBS-LIST\n");
+	(void)proxy_client_worker_output_flush(_worker);
+	return &iter->iter;
+}
+
+static int
+proxy_client_worker_subs_iter_next_line(struct proxy_client_dsync_worker_subs_iter *iter,
+					unsigned int wanted_arg_count,
+					char ***args_r)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)iter->iter.worker;
+	const char *line;
+	char **args;
+	int ret;
+
+	if ((ret = proxy_client_worker_read_line(worker, &line)) <= 0) {
+		if (ret < 0)
+			iter->iter.failed = TRUE;
+		return ret;
+	}
+
+	if ((line[0] == '+' || line[0] == '-') && line[1] == '\0') {
+		/* end of subscribed subscriptions */
+		if (line[0] == '-') {
+			i_error("Worker server's subscription iteration failed");
+			iter->iter.failed = TRUE;
+		}
+		return -1;
+	}
+
+	p_clear(iter->pool);
+	args = p_strsplit(iter->pool, line, "\t");
+	if (str_array_length((const char *const *)args) < wanted_arg_count) {
+		i_error("Invalid subscription input from worker server");
+		iter->iter.failed = TRUE;
+		return -1;
+	}
+	*args_r = args;
+	return 1;
+}
+
+static int
+proxy_client_worker_subs_iter_next(struct dsync_worker_subs_iter *_iter,
+				   struct dsync_worker_subscription *rec_r)
+{
+	struct proxy_client_dsync_worker_subs_iter *iter =
+		(struct proxy_client_dsync_worker_subs_iter *)_iter;
+	char **args;
+	int ret;
+
+	ret = proxy_client_worker_subs_iter_next_line(iter, 4, &args);
+	if (ret <= 0)
+		return ret;
+
+	rec_r->vname = str_tabunescape(args[0]);
+	rec_r->storage_name = str_tabunescape(args[1]);
+	rec_r->ns_prefix = str_tabunescape(args[2]);
+	rec_r->last_change = strtoul(args[3], NULL, 10);
+	return 1;
+}
+
+static int
+proxy_client_worker_subs_iter_next_un(struct dsync_worker_subs_iter *_iter,
+				      struct dsync_worker_unsubscription *rec_r)
+{
+	struct proxy_client_dsync_worker_subs_iter *iter =
+		(struct proxy_client_dsync_worker_subs_iter *)_iter;
+	char **args;
+	int ret;
+
+	ret = proxy_client_worker_subs_iter_next_line(iter, 3, &args);
+	if (ret <= 0)
+		return ret;
+
+	memset(rec_r, 0, sizeof(*rec_r));
+	if (dsync_proxy_mailbox_guid_import(args[0], &rec_r->name_sha1) < 0) {
+		i_error("Invalid subscription input from worker server: "
+			"Invalid unsubscription mailbox GUID");
+		iter->iter.failed = TRUE;
+		return -1;
+	}
+	rec_r->ns_prefix = str_tabunescape(args[1]);
+	rec_r->last_change = strtoul(args[2], NULL, 10);
+	return 1;
+}
+
+static int
+proxy_client_worker_subs_iter_deinit(struct dsync_worker_subs_iter *_iter)
+{
+	struct proxy_client_dsync_worker_subs_iter *iter =
+		(struct proxy_client_dsync_worker_subs_iter *)_iter;
+	int ret = _iter->failed ? -1 : 0;
+
+	pool_unref(&iter->pool);
+	i_free(iter);
+	return ret;
+}
+
+static void
+proxy_client_worker_set_subscribed(struct dsync_worker *_worker,
+				   const char *name, time_t last_change,
+				   bool set)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+
+	T_BEGIN {
+		string_t *str = t_str_new(128);
+
+		str_append(str, "SUBS-SET\t");
+		str_tabescape_write(str, name);
+		str_printfa(str, "\t%s\t%d\n", dec2str(last_change),
+			    set ? 1 : 0);
+		o_stream_send(worker->output, str_data(str), str_len(str));
+	} T_END;
+}
+
+struct proxy_client_dsync_worker_msg_iter {
+	struct dsync_worker_msg_iter iter;
+	pool_t pool;
+	bool done;
+};
+
+static struct dsync_worker_msg_iter *
+proxy_client_worker_msg_iter_init(struct dsync_worker *_worker,
+				  const mailbox_guid_t mailboxes[],
+				  unsigned int mailbox_count)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+	struct proxy_client_dsync_worker_msg_iter *iter;
+	string_t *str;
+	unsigned int i;
+
+	iter = i_new(struct proxy_client_dsync_worker_msg_iter, 1);
+	iter->iter.worker = _worker;
+	iter->pool = pool_alloconly_create("proxy message iter", 10240);
+
+	str = str_new(iter->pool, 512);
+	str_append(str, "MSG-LIST");
+	for (i = 0; i < mailbox_count; i++) T_BEGIN {
+		str_append_c(str, '\t');
+		dsync_proxy_mailbox_guid_export(str, &mailboxes[i]);
+	} T_END;
+	str_append_c(str, '\n');
+	o_stream_send(worker->output, str_data(str), str_len(str));
+	p_clear(iter->pool);
+
+	(void)proxy_client_worker_output_flush(_worker);
+	return &iter->iter;
+}
+
+static int
+proxy_client_worker_msg_iter_next(struct dsync_worker_msg_iter *_iter,
+				  unsigned int *mailbox_idx_r,
+				  struct dsync_message *msg_r)
+{
+	struct proxy_client_dsync_worker_msg_iter *iter =
+		(struct proxy_client_dsync_worker_msg_iter *)_iter;
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_iter->worker;
+	const char *line, *error;
+	int ret;
+
+	if (iter->done)
+		return -1;
+
+	if ((ret = proxy_client_worker_read_line(worker, &line)) <= 0) {
+		if (ret < 0)
+			_iter->failed = TRUE;
+		return ret;
+	}
+
+	if ((line[0] == '+' || line[0] == '-') && line[1] == '\0') {
+		/* end of messages */
+		if (line[0] == '-') {
+			i_error("Worker server's message iteration failed");
+			_iter->failed = TRUE;
+		}
+		iter->done = TRUE;
+		return -1;
+	}
+
+	*mailbox_idx_r = 0;
+	while (*line >= '0' && *line <= '9') {
+		*mailbox_idx_r = *mailbox_idx_r * 10 + (*line - '0');
+		line++;
+	}
+	if (*line != '\t') {
+		i_error("Invalid mailbox idx from worker server");
+		_iter->failed = TRUE;
+		return -1;
+	}
+	line++;
+
+	p_clear(iter->pool);
+	if (dsync_proxy_msg_import(iter->pool, line, msg_r, &error) < 0) {
+		i_error("Invalid message input from worker server: %s", error);
+		_iter->failed = TRUE;
+		return -1;
+	}
+	return 1;
+}
+
+static int
+proxy_client_worker_msg_iter_deinit(struct dsync_worker_msg_iter *_iter)
+{
+	struct proxy_client_dsync_worker_msg_iter *iter =
+		(struct proxy_client_dsync_worker_msg_iter *)_iter;
+	int ret = _iter->failed ? -1 : 0;
+
+	pool_unref(&iter->pool);
+	i_free(iter);
+	return ret;
+}
+
+static void
+proxy_client_worker_cmd(struct proxy_client_dsync_worker *worker, string_t *str)
+{
+	if (worker->save_input == NULL)
+		o_stream_send(worker->output, str_data(str), str_len(str));
+	else
+		str_append_str(worker->pending_commands, str);
+}
+
+static void
+proxy_client_worker_create_mailbox(struct dsync_worker *_worker,
+				   const struct dsync_mailbox *dsync_box)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+
+	T_BEGIN {
+		string_t *str = t_str_new(128);
+
+		str_append(str, "BOX-CREATE\t");
+		dsync_proxy_mailbox_export(str, dsync_box);
+		str_append_c(str, '\n');
+		proxy_client_worker_cmd(worker, str);
+	} T_END;
+}
+
+static void
+proxy_client_worker_delete_mailbox(struct dsync_worker *_worker,
+				   const struct dsync_mailbox *dsync_box)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+
+	T_BEGIN {
+		string_t *str = t_str_new(128);
+
+		str_append(str, "BOX-DELETE\t");
+		dsync_proxy_mailbox_guid_export(str, &dsync_box->mailbox_guid);
+		str_printfa(str, "\t%s\n", dec2str(dsync_box->last_change));
+		proxy_client_worker_cmd(worker, str);
+	} T_END;
+}
+
+static void
+proxy_client_worker_delete_dir(struct dsync_worker *_worker,
+			       const struct dsync_mailbox *dsync_box)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+
+	T_BEGIN {
+		string_t *str = t_str_new(128);
+
+		str_append(str, "DIR-DELETE\t");
+		str_tabescape_write(str, dsync_box->name);
+		str_printfa(str, "\t%s\n", dec2str(dsync_box->last_change));
+		proxy_client_worker_cmd(worker, str);
+	} T_END;
+}
+
+static void
+proxy_client_worker_rename_mailbox(struct dsync_worker *_worker,
+				   const mailbox_guid_t *mailbox,
+				   const struct dsync_mailbox *dsync_box)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+	char sep[2];
+
+	T_BEGIN {
+		string_t *str = t_str_new(128);
+
+		str_append(str, "BOX-RENAME\t");
+		dsync_proxy_mailbox_guid_export(str, mailbox);
+		str_append_c(str, '\t');
+		str_tabescape_write(str, dsync_box->name);
+		str_append_c(str, '\t');
+		sep[0] = dsync_box->name_sep; sep[1] = '\0';
+		str_tabescape_write(str, sep);
+		str_append_c(str, '\n');
+		proxy_client_worker_cmd(worker, str);
+	} T_END;
+}
+
+static void
+proxy_client_worker_update_mailbox(struct dsync_worker *_worker,
+				   const struct dsync_mailbox *dsync_box)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+
+	T_BEGIN {
+		string_t *str = t_str_new(128);
+
+		str_append(str, "BOX-UPDATE\t");
+		dsync_proxy_mailbox_export(str, dsync_box);
+		str_append_c(str, '\n');
+		proxy_client_worker_cmd(worker, str);
+	} T_END;
+}
+
+static void
+proxy_client_worker_select_mailbox(struct dsync_worker *_worker,
+				   const mailbox_guid_t *mailbox,
+				   const ARRAY_TYPE(mailbox_cache_field) *cache_fields)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+
+	if (dsync_guid_equals(&worker->selected_box_guid, mailbox))
+		return;
+	worker->selected_box_guid = *mailbox;
+
+	T_BEGIN {
+		string_t *str = t_str_new(128);
+
+		str_append(str, "BOX-SELECT\t");
+		dsync_proxy_mailbox_guid_export(str, mailbox);
+		if (cache_fields != NULL)
+			dsync_proxy_cache_fields_export(str, cache_fields);
+		str_append_c(str, '\n');
+		proxy_client_worker_cmd(worker, str);
+	} T_END;
+}
+
+static void
+proxy_client_worker_msg_update_metadata(struct dsync_worker *_worker,
+					const struct dsync_message *msg)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+
+	T_BEGIN {
+		string_t *str = t_str_new(128);
+
+		str_printfa(str, "MSG-UPDATE\t%u\t%llu\t", msg->uid,
+			    (unsigned long long)msg->modseq);
+		imap_write_flags(str, msg->flags, msg->keywords);
+		str_append_c(str, '\n');
+		proxy_client_worker_cmd(worker, str);
+	} T_END;
+}
+
+static void
+proxy_client_worker_msg_update_uid(struct dsync_worker *_worker,
+				   uint32_t old_uid, uint32_t new_uid)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+
+	T_BEGIN {
+		string_t *str = t_str_new(64);
+		str_printfa(str, "MSG-UID-CHANGE\t%u\t%u\n", old_uid, new_uid);
+		proxy_client_worker_cmd(worker, str);
+	} T_END;
+}
+
+static void
+proxy_client_worker_msg_expunge(struct dsync_worker *_worker, uint32_t uid)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+
+	T_BEGIN {
+		string_t *str = t_str_new(64);
+		str_printfa(str, "MSG-EXPUNGE\t%u\n", uid);
+		proxy_client_worker_cmd(worker, str);
+	} T_END;
+}
+
+static void
+proxy_client_worker_msg_copy(struct dsync_worker *_worker,
+			     const mailbox_guid_t *src_mailbox,
+			     uint32_t src_uid,
+			     const struct dsync_message *dest_msg,
+			     dsync_worker_copy_callback_t *callback,
+			     void *context)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+	struct proxy_client_request request;
+
+	T_BEGIN {
+		string_t *str = t_str_new(128);
+
+		str_append(str, "MSG-COPY\t");
+		dsync_proxy_mailbox_guid_export(str, src_mailbox);
+		str_printfa(str, "\t%u\t", src_uid);
+		dsync_proxy_msg_export(str, dest_msg);
+		str_append_c(str, '\n');
+		proxy_client_worker_cmd(worker, str);
+	} T_END;
+
+	memset(&request, 0, sizeof(request));
+	request.type = PROXY_CLIENT_REQUEST_TYPE_COPY;
+	request.callback.copy = callback;
+	request.context = context;
+	request.uid = src_uid;
+	aqueue_append(worker->request_queue, &request);
+}
+
+static void
+proxy_client_send_stream_real(struct proxy_client_dsync_worker *worker)
+{
+	dsync_worker_save_callback_t *callback;
+	void *context;
+	struct istream *input;
+	const unsigned char *data;
+	size_t size;
+	int ret;
+
+	while ((ret = i_stream_read_data(worker->save_input,
+					 &data, &size, 0)) > 0) {
+		dsync_proxy_send_dot_output(worker->output,
+					    &worker->save_input_last_lf,
+					    data, size);
+		i_stream_skip(worker->save_input, size);
+
+		if (worker_is_output_stream_full(worker)) {
+			o_stream_uncork(worker->output);
+			if (worker_is_output_stream_full(worker))
+				return;
+			o_stream_cork(worker->output);
+		}
+	}
+	if (ret == 0) {
+		/* waiting for more input */
+		o_stream_uncork(worker->output);
+		if (worker->save_io == NULL) {
+			int fd = i_stream_get_fd(worker->save_input);
+
+			worker->save_io =
+				io_add(fd, IO_READ,
+				       proxy_client_send_stream, worker);
+		}
+		return;
+	}
+	if (worker->save_io != NULL)
+		io_remove(&worker->save_io);
+	if (worker->save_input->stream_errno != 0) {
+		errno = worker->save_input->stream_errno;
+		i_error("proxy: reading message input failed: %m");
+		o_stream_close(worker->output);
+	} else {
+		i_assert(!i_stream_have_bytes_left(worker->save_input));
+		o_stream_send(worker->output, "\n.\n", 3);
+	}
+
+	callback = worker->save_callback;
+	context = worker->save_context;
+	worker->save_callback = NULL;
+	worker->save_context = NULL;
+
+	/* a bit ugly way to free the stream. the problem is that local worker
+	   has set a destroy callback, which in turn can call our msg_save()
+	   again before the i_stream_unref() is finished. */
+	input = worker->save_input;
+	worker->save_input = NULL;
+	i_stream_unref(&input);
+
+	(void)proxy_client_worker_output_flush(&worker->worker);
+
+	callback(context);
+}
+
+static void proxy_client_send_stream(struct proxy_client_dsync_worker *worker)
+{
+	proxy_client_send_stream_real(worker);
+	timeout_reset(worker->to);
+}
+
+static void
+proxy_client_worker_msg_save(struct dsync_worker *_worker,
+			     const struct dsync_message *msg,
+			     const struct dsync_msg_static_data *data,
+			     dsync_worker_save_callback_t *callback,
+			     void *context)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+
+	T_BEGIN {
+		string_t *str = t_str_new(128);
+
+		str_append(str, "MSG-SAVE\t");
+		dsync_proxy_msg_static_export(str, data);
+		str_append_c(str, '\t');
+		dsync_proxy_msg_export(str, msg);
+		str_append_c(str, '\n');
+		proxy_client_worker_cmd(worker, str);
+	} T_END;
+
+	i_assert(worker->save_input == NULL);
+	worker->save_callback = callback;
+	worker->save_context = context;
+	worker->save_input = data->input;
+	worker->save_input_last_lf = TRUE;
+	i_stream_ref(worker->save_input);
+	proxy_client_send_stream(worker);
+}
+
+static void
+proxy_client_worker_msg_save_cancel(struct dsync_worker *_worker)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+
+	if (worker->save_io != NULL)
+		io_remove(&worker->save_io);
+	if (worker->save_input != NULL)
+		i_stream_unref(&worker->save_input);
+}
+
+static void
+proxy_client_worker_msg_get(struct dsync_worker *_worker,
+			    const mailbox_guid_t *mailbox, uint32_t uid,
+			    dsync_worker_msg_callback_t *callback,
+			    void *context)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+	struct proxy_client_request request;
+
+	T_BEGIN {
+		string_t *str = t_str_new(128);
+
+		str_append(str, "MSG-GET\t");
+		dsync_proxy_mailbox_guid_export(str, mailbox);
+		str_printfa(str, "\t%u\n", uid);
+		proxy_client_worker_cmd(worker, str);
+	} T_END;
+
+	memset(&request, 0, sizeof(request));
+	request.type = PROXY_CLIENT_REQUEST_TYPE_GET;
+	request.callback.get = callback;
+	request.context = context;
+	request.uid = uid;
+	aqueue_append(worker->request_queue, &request);
+}
+
+static void
+proxy_client_worker_finish(struct dsync_worker *_worker,
+			   dsync_worker_finish_callback_t *callback,
+			   void *context)
+{
+	struct proxy_client_dsync_worker *worker =
+		(struct proxy_client_dsync_worker *)_worker;
+	struct proxy_client_request request;
+
+	i_assert(worker->save_input == NULL);
+	i_assert(!worker->finishing);
+
+	worker->finishing = TRUE;
+	worker->finished = FALSE;
+
+	o_stream_send_str(worker->output, "FINISH\n");
+	o_stream_uncork(worker->output);
+
+	memset(&request, 0, sizeof(request));
+	request.type = PROXY_CLIENT_REQUEST_TYPE_FINISH;
+	request.callback.finish = callback;
+	request.context = context;
+	aqueue_append(worker->request_queue, &request);
+}
+
+struct dsync_worker_vfuncs proxy_client_dsync_worker = {
+	proxy_client_worker_deinit,
+
+	proxy_client_worker_is_output_full,
+	proxy_client_worker_output_flush,
+
+	proxy_client_worker_mailbox_iter_init,
+	proxy_client_worker_mailbox_iter_next,
+	proxy_client_worker_mailbox_iter_deinit,
+
+	proxy_client_worker_subs_iter_init,
+	proxy_client_worker_subs_iter_next,
+	proxy_client_worker_subs_iter_next_un,
+	proxy_client_worker_subs_iter_deinit,
+	proxy_client_worker_set_subscribed,
+
+	proxy_client_worker_msg_iter_init,
+	proxy_client_worker_msg_iter_next,
+	proxy_client_worker_msg_iter_deinit,
+
+	proxy_client_worker_create_mailbox,
+	proxy_client_worker_delete_mailbox,
+	proxy_client_worker_delete_dir,
+	proxy_client_worker_rename_mailbox,
+	proxy_client_worker_update_mailbox,
+
+	proxy_client_worker_select_mailbox,
+	proxy_client_worker_msg_update_metadata,
+	proxy_client_worker_msg_update_uid,
+	proxy_client_worker_msg_expunge,
+	proxy_client_worker_msg_copy,
+	proxy_client_worker_msg_save,
+	proxy_client_worker_msg_save_cancel,
+	proxy_client_worker_msg_get,
+	proxy_client_worker_finish
+};