--- a/src/dsync/dsync-proxy-client.c	Mon Oct 25 18:26:50 2010 +0100
+++ b/src/dsync/dsync-proxy-client.c	Mon Oct 25 18:53:04 2010 +0100
@@ -123,23 +123,12 @@
 }
 
 static void
-proxy_client_worker_msg_get_done(struct proxy_client_dsync_worker *worker)
+proxy_client_worker_msg_get_finish(struct proxy_client_dsync_worker *worker)
 {
-	struct istream *input = worker->msg_get_data.input;
-	const unsigned char *data;
-	size_t size;
-
-	i_assert(worker->io == NULL);
-
 	worker->msg_get_data.input = NULL;
 	worker->io = io_add(worker->fd_in, IO_READ,
 			    proxy_client_worker_input, worker);
 
-	/* we'll need to read the input until EOF or we'll start treating the
-	   input as commands. make sure saving read everything. */
-	while ((i_stream_read_data(input, &data, &size, 0)) > 0)
-		i_stream_skip(input, size);
-
 	/* 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) */
@@ -149,6 +138,43 @@
 	}
 }
 
+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);
+	}
+}
+
+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,
@@ -232,7 +258,7 @@
 	}
 
 	request->callback.get(result, &worker->msg_get_data, request->context);
-	return worker->io != NULL;
+	return worker->io != NULL && worker->msg_get_data.input == NULL;
 }
 
 static void

--- a/src/dsync/dsync-proxy-server.c	Mon Oct 25 18:26:50 2010 +0100
+++ b/src/dsync/dsync-proxy-server.c	Mon Oct 25 18:53:04 2010 +0100
@@ -192,7 +192,8 @@
 		i_stream_unref(&server->get_input);
 	pool_unref(&server->cmd_pool);
 	timeout_remove(&server->to);
-	io_remove(&server->io);
+	if (server->io != NULL)
+		io_remove(&server->io);
 	i_stream_destroy(&server->input);
 	o_stream_destroy(&server->output);
 	if (close(server->fd_in) < 0)

--- a/src/dsync/dsync-worker-local.c	Mon Oct 25 18:26:50 2010 +0100
+++ b/src/dsync/dsync-worker-local.c	Mon Oct 25 18:53:04 2010 +0100
@@ -1706,6 +1706,7 @@
 		i_error("Can't save message to mailbox %s: %s",
 			mailbox_get_vname(dest_box),
 			mail_storage_get_last_error(storage, NULL));
+		mailbox_save_cancel(&save_ctx);
 		dsync_worker_set_failure(_worker);
 		callback(context);
 		return;