<commit_before>/* Copyright (c) 2009-2010 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;

	i_warning("save done");
	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(&->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;

		i_warning("retr %d", save_ctx->msg->uid);
		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;
	}
	return dsync_worker_is_output_full(dest_iter->worker) ? 0 : 1;
}
<commit_msg>dsync: Fixed accidentally committed broken code.<commit_after>/* Copyright (c) 2009-2010 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;
	}
	return dsync_worker_is_output_full(dest_iter->worker) ? 0 : 1;
}

{'fraction_non_alphanumeric': 0.09879146727228486, 'fraction_numerical': 0.004217698110146808, 'mean_word_length': 4.386631716906947, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 177, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1763305', 'n_tokens_mistral': 4637, 'n_tokens_neox': 4260, 'n_words': 1106}