xref: /dports/mail/dovecot/dovecot-2.3.17/src/lib-index/mail-cache-transaction.c

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

#include "lib.h"
#include "ioloop.h"
#include "array.h"
#include "buffer.h"
#include "module-context.h"
#include "file-cache.h"
#include "file-set-size.h"
#include "read-full.h"
#include "write-full.h"
#include "mail-cache-private.h"
#include "mail-index-transaction-private.h"

#include <stddef.h>
#include <sys/stat.h>

#define MAIL_CACHE_INIT_WRITE_BUFFER (1024*16)

#define CACHE_TRANS_CONTEXT(obj) \
	MODULE_CONTEXT(obj, cache_mail_index_transaction_module)
#define CACHE_TRANS_CONTEXT_REQUIRE(obj) \
	MODULE_CONTEXT_REQUIRE(obj, cache_mail_index_transaction_module)

struct mail_cache_transaction_rec {
	uint32_t seq;
	uint32_t cache_data_pos;
};

struct mail_cache_transaction_ctx {
	union mail_index_transaction_module_context module_ctx;
	struct mail_index_transaction_vfuncs super;

	struct mail_cache *cache;
	struct mail_cache_view *view;
	struct mail_index_transaction *trans;

	uint32_t cache_file_seq;
	uint32_t first_new_seq;

	buffer_t *cache_data;
	ARRAY(uint8_t) cache_field_idx_used;
	ARRAY(struct mail_cache_transaction_rec) cache_data_seq;
	ARRAY_TYPE(seq_range) cache_data_wanted_seqs;
	uint32_t prev_seq, min_seq;
	size_t last_rec_pos;

	unsigned int records_written;

	bool tried_purging:1;
	bool decisions_refreshed:1;
	bool have_noncommited_mails:1;
	bool changes:1;
};

static MODULE_CONTEXT_DEFINE_INIT(cache_mail_index_transaction_module,
				  &mail_index_module_register);

static int mail_cache_transaction_lock(struct mail_cache_transaction_ctx *ctx);
static bool
mail_cache_transaction_update_last_rec_size(struct mail_cache_transaction_ctx *ctx,
					    size_t *size_r);
static int mail_cache_header_rewrite_fields(struct mail_cache *cache);

static void mail_index_transaction_cache_reset(struct mail_index_transaction *t)
{
	struct mail_cache_transaction_ctx *ctx = CACHE_TRANS_CONTEXT_REQUIRE(t);
	struct mail_index_transaction_vfuncs super = ctx->super;

	mail_cache_transaction_reset(ctx);
	super.reset(t);
}

static int
mail_index_transaction_cache_commit(struct mail_index_transaction *t,
				    struct mail_index_transaction_commit_result *result_r)
{
	struct mail_cache_transaction_ctx *ctx = CACHE_TRANS_CONTEXT_REQUIRE(t);
	struct mail_index_transaction_vfuncs super = ctx->super;

	/* a failed cache commit isn't important enough to fail the entire
	   index transaction, so we'll just ignore it */
	(void)mail_cache_transaction_commit(&ctx);
	return super.commit(t, result_r);
}

static void
mail_index_transaction_cache_rollback(struct mail_index_transaction *t)
{
	struct mail_cache_transaction_ctx *ctx = CACHE_TRANS_CONTEXT_REQUIRE(t);
	struct mail_index_transaction_vfuncs super = ctx->super;

	mail_cache_transaction_rollback(&ctx);
	super.rollback(t);
}

struct mail_cache_transaction_ctx *
mail_cache_get_transaction(struct mail_cache_view *view,
			   struct mail_index_transaction *t)
{
	struct mail_cache_transaction_ctx *ctx;

	ctx = !cache_mail_index_transaction_module.id.module_id_set ? NULL :
		CACHE_TRANS_CONTEXT(t);

	if (ctx != NULL)
		return ctx;

	ctx = i_new(struct mail_cache_transaction_ctx, 1);
	ctx->cache = view->cache;
	ctx->view = view;
	ctx->trans = t;

	i_assert(view->transaction == NULL);
	view->transaction = ctx;
	view->trans_view = mail_index_transaction_open_updated_view(t);

	ctx->super = t->v;
	t->v.reset = mail_index_transaction_cache_reset;
	t->v.commit = mail_index_transaction_cache_commit;
	t->v.rollback = mail_index_transaction_cache_rollback;

	MODULE_CONTEXT_SET(t, cache_mail_index_transaction_module, ctx);
	return ctx;
}

static void
mail_cache_transaction_forget_flushed(struct mail_cache_transaction_ctx *ctx,
				      bool reset_id_changed)
{
	uint32_t new_cache_file_seq = MAIL_CACHE_IS_UNUSABLE(ctx->cache) ? 0 :
		ctx->cache->hdr->file_seq;
	if (reset_id_changed && ctx->records_written > 0) {
		e_warning(ctx->cache->event,
			  "Purging lost %u written cache records "
			  "(reset_id changed %u -> %u)", ctx->records_written,
			  ctx->cache_file_seq, new_cache_file_seq);
		/* don't increase deleted_record_count in the new file */
		ctx->records_written = 0;
	}
	ctx->cache_file_seq = new_cache_file_seq;
	/* forget all cache extension updates even if reset_id doesn't change */
	mail_index_ext_set_reset_id(ctx->trans, ctx->cache->ext_id,
				    ctx->cache_file_seq);
}

void mail_cache_transaction_reset(struct mail_cache_transaction_ctx *ctx)
{
	mail_cache_transaction_forget_flushed(ctx, FALSE);
	if (ctx->cache_data != NULL)
		buffer_set_used_size(ctx->cache_data, 0);
	if (array_is_created(&ctx->cache_data_seq))
		array_clear(&ctx->cache_data_seq);
	ctx->prev_seq = 0;
	ctx->last_rec_pos = 0;

	ctx->changes = FALSE;
}

void mail_cache_transaction_rollback(struct mail_cache_transaction_ctx **_ctx)
{
	struct mail_cache_transaction_ctx *ctx = *_ctx;

	*_ctx = NULL;

	if (ctx->records_written > 0) {
		/* we already wrote to the cache file. we can't (or don't want
		   to) delete that data, so just mark it as deleted space */
		if (mail_cache_transaction_lock(ctx) > 0) {
			ctx->cache->hdr_copy.deleted_record_count +=
				ctx->records_written;
			ctx->cache->hdr_modified = TRUE;
			(void)mail_cache_flush_and_unlock(ctx->cache);
		}
	}

	MODULE_CONTEXT_UNSET(ctx->trans, cache_mail_index_transaction_module);

	ctx->view->transaction = NULL;
	ctx->view->trans_seq1 = ctx->view->trans_seq2 = 0;

	mail_index_view_close(&ctx->view->trans_view);
	buffer_free(&ctx->cache_data);
	if (array_is_created(&ctx->cache_data_seq))
		array_free(&ctx->cache_data_seq);
	if (array_is_created(&ctx->cache_data_wanted_seqs))
		array_free(&ctx->cache_data_wanted_seqs);
	array_free(&ctx->cache_field_idx_used);
	i_free(ctx);
}

bool mail_cache_transactions_have_changes(struct mail_cache *cache)
{
	struct mail_cache_view *view;

	for (view = cache->views; view != NULL; view = view->next) {
		if (view->transaction != NULL &&
		    view->transaction->changes)
			return TRUE;
	}
	return FALSE;
}

static int
mail_cache_transaction_purge(struct mail_cache_transaction_ctx *ctx,
			     const char *reason)
{
	struct mail_cache *cache = ctx->cache;

	ctx->tried_purging = TRUE;

	uint32_t purge_file_seq =
		MAIL_CACHE_IS_UNUSABLE(cache) ? 0 : cache->hdr->file_seq;

	int ret = mail_cache_purge(cache, purge_file_seq, reason);
	/* already written cache records must be forgotten, but records in
	   memory can still be written to the new cache file */
	mail_cache_transaction_forget_flushed(ctx, TRUE);
	return ret;
}

static int mail_cache_transaction_lock(struct mail_cache_transaction_ctx *ctx)
{
	struct mail_cache *cache = ctx->cache;
	const uoff_t cache_max_size =
		cache->index->optimization_set.cache.max_size;
	int ret;

	if ((ret = mail_cache_lock(cache)) <= 0) {
		if (ret < 0)
			return -1;

		if (!ctx->tried_purging) {
			if (mail_cache_transaction_purge(ctx, "creating cache") < 0)
				return -1;
			return mail_cache_transaction_lock(ctx);
		} else {
			return 0;
		}
	}
	i_assert(!MAIL_CACHE_IS_UNUSABLE(cache));

	if (!ctx->tried_purging && ctx->cache_data != NULL &&
	    cache->last_stat_size + ctx->cache_data->used > cache_max_size) {
		/* Looks like cache file is becoming too large. Try to purge
		   it to free up some space. */
		if (cache->hdr->continued_record_count > 0 ||
		    cache->hdr->deleted_record_count > 0) {
			mail_cache_unlock(cache);
			(void)mail_cache_transaction_purge(ctx, "cache is too large");
			return mail_cache_transaction_lock(ctx);
		}
	}

	if (ctx->cache_file_seq == 0)
		ctx->cache_file_seq = cache->hdr->file_seq;
	else if (ctx->cache_file_seq != cache->hdr->file_seq) {
		/* already written cache records must be forgotten, but records
		   in memory can still be written to the new cache file */
		mail_cache_transaction_forget_flushed(ctx, TRUE);
		i_assert(ctx->cache_file_seq == cache->hdr->file_seq);
	}
	return 1;
}

const struct mail_cache_record *
mail_cache_transaction_lookup_rec(struct mail_cache_transaction_ctx *ctx,
				  unsigned int seq,
				  unsigned int *trans_next_idx)
{
	const struct mail_cache_transaction_rec *recs;
	unsigned int i, count;

	recs = array_get(&ctx->cache_data_seq, &count);
	for (i = *trans_next_idx; i < count; i++) {
		if (recs[i].seq == seq) {
			*trans_next_idx = i + 1;
			return CONST_PTR_OFFSET(ctx->cache_data->data,
						recs[i].cache_data_pos);
		}
	}
	*trans_next_idx = i + 1;
	if (seq == ctx->prev_seq && i == count) {
		/* update the unfinished record's (temporary) size and
		   return it */
		size_t size;
		if (!mail_cache_transaction_update_last_rec_size(ctx, &size))
			return NULL;
		return CONST_PTR_OFFSET(ctx->cache_data->data,
					ctx->last_rec_pos);
	}
	return NULL;
}

static void
mail_cache_transaction_update_index(struct mail_cache_transaction_ctx *ctx,
				    uint32_t write_offset, bool committing)
{
	struct mail_cache *cache = ctx->cache;
	struct mail_index_transaction *trans;
	const struct mail_cache_record *rec = ctx->cache_data->data;
	const struct mail_cache_transaction_rec *recs;
	uint32_t i, seq_count;

	if (committing) {
		/* The transaction is being committed now. Use it. */
		trans = ctx->trans;
	} else if (ctx->have_noncommited_mails) {
		/* Some of the mails haven't been committed yet. We must use
		   the provided transaction to update the cache records. */
		trans = ctx->trans;
	} else {
		/* We can commit these changes immediately. This way even if
		   the provided transaction runs for a very long time, we
		   still once in a while commit the cache changes so they
		   become visible to other processes as well. */
		trans = mail_index_transaction_begin(ctx->view->trans_view,
			MAIL_INDEX_TRANSACTION_FLAG_EXTERNAL);
	}

	mail_index_ext_using_reset_id(trans, ctx->cache->ext_id,
				      ctx->cache_file_seq);

	/* write the cache_offsets to index file. records' prev_offset
	   is updated to point to old cache record when index is being
	   synced. */
	recs = array_get(&ctx->cache_data_seq, &seq_count);
	for (i = 0; i < seq_count; i++) {
		mail_index_update_ext(trans, recs[i].seq, cache->ext_id,
				      &write_offset, NULL);

		write_offset += rec->size;
		rec = CONST_PTR_OFFSET(rec, rec->size);
		ctx->records_written++;
	}
	if (trans != ctx->trans) {
		i_assert(cache->index->log_sync_locked);
		if (mail_index_transaction_commit(&trans) < 0) {
			/* failed, but can't really do anything */
		} else {
			ctx->records_written = 0;
		}
	}
}

static int
mail_cache_link_records(struct mail_cache_transaction_ctx *ctx,
			uint32_t write_offset)
{
	struct mail_index_map *map;
	struct mail_cache_record *rec;
	const struct mail_cache_transaction_rec *recs;
	const uint32_t *prev_offsetp;
	ARRAY_TYPE(uint32_t) seq_offsets;
	uint32_t i, seq_count, reset_id, prev_offset, *offsetp;
	const void *data;

	i_assert(ctx->min_seq != 0);

	i_array_init(&seq_offsets, 64);
	recs = array_get(&ctx->cache_data_seq, &seq_count);
	rec = buffer_get_modifiable_data(ctx->cache_data, NULL);
	for (i = 0; i < seq_count; i++) {
		offsetp = array_idx_get_space(&seq_offsets,
					       recs[i].seq - ctx->min_seq);
		if (*offsetp != 0)
			prev_offset = *offsetp;
		else {
			mail_index_lookup_ext_full(ctx->view->trans_view, recs[i].seq,
						   ctx->cache->ext_id, &map,
						   &data, NULL);
			prev_offsetp = data;

			if (prev_offsetp == NULL || *prev_offsetp == 0)
				prev_offset = 0;
			else if (mail_index_ext_get_reset_id(ctx->view->trans_view, map,
							     ctx->cache->ext_id,
							     &reset_id) &&
				 reset_id == ctx->cache_file_seq)
				prev_offset = *prev_offsetp;
			else
				prev_offset = 0;
			if (prev_offset >= write_offset) {
				mail_cache_set_corrupted(ctx->cache,
					"Cache record offset points outside existing file");
				array_free(&seq_offsets);
				return -1;
			}
		}

		if (prev_offset != 0) {
			/* link this record to previous one */
			rec->prev_offset = prev_offset;
			ctx->cache->hdr_copy.continued_record_count++;
		} else {
			ctx->cache->hdr_copy.record_count++;
		}
		*offsetp = write_offset;

		write_offset += rec->size;
		rec = PTR_OFFSET(rec, rec->size);
	}
	array_free(&seq_offsets);
	ctx->cache->hdr_modified = TRUE;
	return 0;
}

static bool
mail_cache_transaction_set_used(struct mail_cache_transaction_ctx *ctx)
{
	const uint8_t *cache_fields_used;
	unsigned int field_idx, count;
	bool missing_file_fields = FALSE;

	cache_fields_used = array_get(&ctx->cache_field_idx_used, &count);
	i_assert(count <= ctx->cache->fields_count);
	for (field_idx = 0; field_idx < count; field_idx++) {
		if (cache_fields_used[field_idx] != 0) {
			ctx->cache->fields[field_idx].used = TRUE;
			if (ctx->cache->field_file_map[field_idx] == (uint32_t)-1)
				missing_file_fields = TRUE;
		}
	}
	return missing_file_fields;
}

static int
mail_cache_transaction_update_fields(struct mail_cache_transaction_ctx *ctx)
{
	unsigned char *p;
	const unsigned char *end, *rec_end;
	uint32_t field_idx, data_size;

	if (mail_cache_transaction_set_used(ctx)) {
		/* add missing fields to cache */
		if (mail_cache_header_rewrite_fields(ctx->cache) < 0)
			return -1;
		/* make sure they were actually added */
		if (mail_cache_transaction_set_used(ctx)) {
			mail_index_set_error(ctx->cache->index,
				"Cache file %s: Unexpectedly lost newly added field",
				ctx->cache->filepath);
			return -1;
		}
	}

	/* Go through all the added cache records and replace the in-memory
	   field_idx with the cache file-specific field index. Update only
	   up to last_rec_pos, because that's how far flushing is done. The
	   fields after that keep the in-memory field_idx until the next
	   flush. */
	p = buffer_get_modifiable_data(ctx->cache_data, NULL);
	end = CONST_PTR_OFFSET(ctx->cache_data->data, ctx->last_rec_pos);
	rec_end = p;
	while (p < end) {
		if (p >= rec_end) {
			/* next cache record */
			i_assert(p == rec_end);
			const struct mail_cache_record *rec =
				(const struct mail_cache_record *)p;
			/* note that the last rec->size==0 */
			rec_end = CONST_PTR_OFFSET(p, rec->size);
			p += sizeof(*rec);
		}
		/* replace field_idx */
		uint32_t *file_fieldp = (uint32_t *)p;
		field_idx = *file_fieldp;
		*file_fieldp = ctx->cache->field_file_map[field_idx];
		i_assert(*file_fieldp != (uint32_t)-1);
		p += sizeof(field_idx);

		/* Skip to next cache field. Next is <data size> if the field
		   is not fixed size. */
		data_size = ctx->cache->fields[field_idx].field.field_size;
		if (data_size == UINT_MAX) {
			memcpy(&data_size, p, sizeof(data_size));
			p += sizeof(data_size);
		}
		/* data & 32bit padding */
		p += data_size;
		if ((data_size & 3) != 0)
			p += 4 - (data_size & 3);
	}
	i_assert(p == end);
	return 0;
}

static void
mail_cache_transaction_drop_last_flush(struct mail_cache_transaction_ctx *ctx)
{
	buffer_copy(ctx->cache_data, 0,
		    ctx->cache_data, ctx->last_rec_pos, SIZE_MAX);
	buffer_set_used_size(ctx->cache_data,
			     ctx->cache_data->used - ctx->last_rec_pos);
	ctx->last_rec_pos = 0;
	ctx->min_seq = 0;

	array_clear(&ctx->cache_data_seq);
	array_clear(&ctx->cache_data_wanted_seqs);
}

static int
mail_cache_transaction_flush(struct mail_cache_transaction_ctx *ctx,
			     bool committing)
{
	struct stat st;
	uint32_t write_offset = 0;
	int ret = 0;

	i_assert(!ctx->cache->locked);

	if (array_count(&ctx->cache_data_seq) == 0) {
		/* we had done some changes, but they were aborted. */
		i_assert(ctx->last_rec_pos == 0);
		ctx->min_seq = 0;
		return 0;
	}

	/* If we're going to be committing a transaction, the log must be
	   locked before we lock cache or we can deadlock. */
	bool lock_log = !ctx->cache->index->log_sync_locked &&
		!committing && !ctx->have_noncommited_mails;
	if (lock_log) {
		uint32_t file_seq;
		uoff_t file_offset;

		if (mail_transaction_log_sync_lock(ctx->cache->index->log,
				"mail cache transaction flush",
				&file_seq, &file_offset) < 0)
			return -1;
	}

	if (mail_cache_transaction_lock(ctx) <= 0) {
		if (lock_log) {
			mail_transaction_log_sync_unlock(ctx->cache->index->log,
				"mail cache transaction flush: cache lock failed");
		}
		return -1;
	}

	i_assert(ctx->cache_data != NULL);
	i_assert(ctx->last_rec_pos <= ctx->cache_data->used);

	if (mail_cache_transaction_update_fields(ctx) < 0) {
		if (lock_log) {
			mail_transaction_log_sync_unlock(ctx->cache->index->log,
				"mail cache transaction flush: field update failed");
		}
		mail_cache_unlock(ctx->cache);
		return -1;
	}

	/* we need to get the final write offset for linking records */
	if (fstat(ctx->cache->fd, &st) < 0) {
		if (!ESTALE_FSTAT(errno))
			mail_cache_set_syscall_error(ctx->cache, "fstat()");
		ret = -1;
	} else if ((uoff_t)st.st_size + ctx->last_rec_pos > ctx->cache->index->optimization_set.cache.max_size) {
		mail_cache_set_corrupted(ctx->cache, "Cache file too large");
		ret = -1;
	} else {
		write_offset = st.st_size;
		if (mail_cache_link_records(ctx, write_offset) < 0)
			ret = -1;
	}

	/* write to cache file */
	if (ret < 0 ||
	    mail_cache_append(ctx->cache, ctx->cache_data->data,
			      ctx->last_rec_pos, &write_offset) < 0)
		ret = -1;
	else {
		/* update records' cache offsets to index */
		mail_cache_transaction_update_index(ctx, write_offset,
						    committing);
	}
	if (mail_cache_flush_and_unlock(ctx->cache) < 0)
		ret = -1;

	if (lock_log) {
		mail_transaction_log_sync_unlock(ctx->cache->index->log,
			"mail cache transaction flush");
	}
	return ret;
}

static void
mail_cache_transaction_drop_unwanted(struct mail_cache_transaction_ctx *ctx,
				     size_t space_needed)
{
	struct mail_cache_transaction_rec *recs;
	unsigned int i, count;

	recs = array_get_modifiable(&ctx->cache_data_seq, &count);
	/* find out how many records to delete. delete all unwanted sequences,
	   and if that's not enough delete some more. */
	for (i = 0; i < count; i++) {
		if (seq_range_exists(&ctx->cache_data_wanted_seqs, recs[i].seq)) {
			if (recs[i].cache_data_pos >= space_needed)
				break;
			/* we're going to forcibly delete it - remove it also
			   from the array since it's no longer useful there */
			seq_range_array_remove(&ctx->cache_data_wanted_seqs,
					       recs[i].seq);
		}
	}
	unsigned int deleted_count = i;
	size_t deleted_space = i < count ?
		recs[i].cache_data_pos : ctx->last_rec_pos;
	for (; i < count; i++)
		recs[i].cache_data_pos -= deleted_space;
	ctx->last_rec_pos -= deleted_space;
	array_delete(&ctx->cache_data_seq, 0, deleted_count);
	buffer_delete(ctx->cache_data, 0, deleted_space);
}

static bool
mail_cache_transaction_update_last_rec_size(struct mail_cache_transaction_ctx *ctx,
					    size_t *size_r)
{
	struct mail_cache_record *rec;
	void *data;
	size_t size;

	data = buffer_get_modifiable_data(ctx->cache_data, &size);
	rec = PTR_OFFSET(data, ctx->last_rec_pos);
	rec->size = size - ctx->last_rec_pos;
	if (rec->size == sizeof(*rec))
		return FALSE;
	i_assert(rec->size > sizeof(*rec));
	*size_r = rec->size;
	return TRUE;
}

static void
mail_cache_transaction_update_last_rec(struct mail_cache_transaction_ctx *ctx)
{
	struct mail_cache_transaction_rec *trans_rec;
	size_t size;

	if (!mail_cache_transaction_update_last_rec_size(ctx, &size) ||
	    size > ctx->cache->index->optimization_set.cache.record_max_size) {
		buffer_set_used_size(ctx->cache_data, ctx->last_rec_pos);
		return;
	}

	if (ctx->min_seq > ctx->prev_seq || ctx->min_seq == 0)
		ctx->min_seq = ctx->prev_seq;
	trans_rec = array_append_space(&ctx->cache_data_seq);
	trans_rec->seq = ctx->prev_seq;
	trans_rec->cache_data_pos = ctx->last_rec_pos;
	ctx->last_rec_pos = ctx->cache_data->used;
}

static void
mail_cache_transaction_switch_seq(struct mail_cache_transaction_ctx *ctx)
{
	struct mail_cache_record new_rec;

	if (ctx->prev_seq != 0) {
		/* update previously added cache record's size */
		mail_cache_transaction_update_last_rec(ctx);
	} else if (ctx->cache_data == NULL) {
		ctx->cache_data =
			buffer_create_dynamic(default_pool,
					      MAIL_CACHE_INIT_WRITE_BUFFER);
		i_array_init(&ctx->cache_data_seq, 64);
		i_array_init(&ctx->cache_data_wanted_seqs, 32);
		i_array_init(&ctx->cache_field_idx_used, 64);
	}

	i_zero(&new_rec);
	buffer_append(ctx->cache_data, &new_rec, sizeof(new_rec));

	ctx->prev_seq = 0;
	ctx->changes = TRUE;
}

int mail_cache_transaction_commit(struct mail_cache_transaction_ctx **_ctx)
{
	struct mail_cache_transaction_ctx *ctx = *_ctx;
	int ret = 0;

	if (ctx->changes) {
		if (ctx->prev_seq != 0)
			mail_cache_transaction_update_last_rec(ctx);
		if (mail_cache_transaction_flush(ctx, TRUE) < 0)
			ret = -1;
		else {
			/* successfully wrote everything */
			ctx->records_written = 0;
		}
		/* Here would be a good place to do fdatasync() to make sure
		   everything is written before offsets are updated to index.
		   However it slows down I/O needlessly and we're pretty good
		   at catching and fixing cache corruption, so we no longer do
		   it. */
	}
	mail_cache_transaction_rollback(_ctx);
	return ret;
}

static int
mail_cache_header_fields_write(struct mail_cache *cache, const buffer_t *buffer)
{
	uint32_t offset, hdr_offset;

	i_assert(cache->locked);

	offset = 0;
	if (mail_cache_append(cache, buffer->data, buffer->used, &offset) < 0)
		return -1;

	if (cache->index->set.fsync_mode == FSYNC_MODE_ALWAYS) {
		if (fdatasync(cache->fd) < 0) {
			mail_cache_set_syscall_error(cache, "fdatasync()");
			return -1;
		}
	}
	/* find offset to the previous header's "next_offset" field */
	if (mail_cache_header_fields_get_next_offset(cache, &hdr_offset) < 0)
		return -1;

	/* update the next_offset offset, so our new header will be found */
	offset = mail_index_uint32_to_offset(offset);
	if (mail_cache_write(cache, &offset, sizeof(offset), hdr_offset) < 0)
		return -1;

	if (hdr_offset == offsetof(struct mail_cache_header,
				   field_header_offset)) {
		/* we're adding the first field. hdr_copy needs to be kept
		   in sync so unlocking won't overwrite it. */
		cache->hdr_copy.field_header_offset = hdr_offset;
		cache->hdr_ro_copy.field_header_offset = hdr_offset;
	}
	return 0;
}

static int mail_cache_header_rewrite_fields(struct mail_cache *cache)
{
	int ret;

	/* re-read header to make sure we don't lose any fields. */
	if (mail_cache_header_fields_read(cache) < 0)
		return -1;

	T_BEGIN {
		buffer_t *buffer;

		buffer = t_buffer_create(256);
		mail_cache_header_fields_get(cache, buffer);
		ret = mail_cache_header_fields_write(cache, buffer);
	} T_END;

	if (ret == 0) {
		/* we wrote all the headers, so there are no pending changes */
		cache->field_header_write_pending = FALSE;
		ret = mail_cache_header_fields_read(cache);
	}
	return ret;
}

static void
mail_cache_transaction_refresh_decisions(struct mail_cache_transaction_ctx *ctx)
{
	if (ctx->decisions_refreshed)
		return;

	/* Read latest caching decisions from the cache file's header once
	   per transaction. */
	if (!ctx->cache->opened)
		(void)mail_cache_open_and_verify(ctx->cache);
	else
		(void)mail_cache_header_fields_read(ctx->cache);
	ctx->decisions_refreshed = TRUE;
}

void mail_cache_add(struct mail_cache_transaction_ctx *ctx, uint32_t seq,
		    unsigned int field_idx, const void *data, size_t data_size)
{
	uint32_t data_size32;
	unsigned int fixed_size;
	size_t full_size, record_size;

	i_assert(field_idx < ctx->cache->fields_count);
	i_assert(data_size < (uint32_t)-1);

	if (ctx->cache->fields[field_idx].field.decision ==
	    (MAIL_CACHE_DECISION_NO | MAIL_CACHE_DECISION_FORCED))
		return;

	if (seq >= ctx->trans->first_new_seq)
		ctx->have_noncommited_mails = TRUE;

	/* If the cache file exists, make sure the caching decisions have been
	   read. */
	mail_cache_transaction_refresh_decisions(ctx);

	mail_cache_decision_add(ctx->view, seq, field_idx);

	fixed_size = ctx->cache->fields[field_idx].field.field_size;
	i_assert(fixed_size == UINT_MAX || fixed_size == data_size);

	data_size32 = (uint32_t)data_size;
	full_size = sizeof(field_idx) + ((data_size + 3) & ~3U);
	if (fixed_size == UINT_MAX)
		full_size += sizeof(data_size32);

	if (ctx->prev_seq != seq) {
		mail_cache_transaction_switch_seq(ctx);
		ctx->prev_seq = seq;
		seq_range_array_add(&ctx->cache_data_wanted_seqs, seq);

		/* remember roughly what we have modified, so cache lookups can
		   look into transactions to see changes. */
		if (seq < ctx->view->trans_seq1 || ctx->view->trans_seq1 == 0)
			ctx->view->trans_seq1 = seq;
		if (seq > ctx->view->trans_seq2)
			ctx->view->trans_seq2 = seq;
	}

	if (mail_cache_transaction_update_last_rec_size(ctx, &record_size) &&
	    record_size + full_size >
	    ctx->cache->index->optimization_set.cache.record_max_size) {
		/* Adding this field would exceed the cache record's maximum
		   size. If we don't add this, it's possible that other fields
		   could still be added. */
		return;
	}

	/* Remember that this field has been used within the transaction. Later
	   on we fill mail_cache_field_private.used with it. We can't rely on
	   setting it here, because cache purging may run and clear it. */
	uint8_t field_idx_set = 1;
	array_idx_set(&ctx->cache_field_idx_used, field_idx, &field_idx_set);

	/* Remember that this value exists for the mail, in case we try to look
	   it up. Note that this gets forgotten whenever changing the mail. */
	buffer_write(ctx->view->cached_exists_buf, field_idx,
		     &ctx->view->cached_exists_value, 1);

	if (ctx->cache_data->used + full_size > MAIL_CACHE_MAX_WRITE_BUFFER &&
	    ctx->last_rec_pos > 0) {
		/* time to flush our buffer. */
		if (MAIL_INDEX_IS_IN_MEMORY(ctx->cache->index)) {
			/* just drop the old data to free up memory */
			size_t space_needed = ctx->cache_data->used +
				full_size - MAIL_CACHE_MAX_WRITE_BUFFER;
			mail_cache_transaction_drop_unwanted(ctx, space_needed);
		} else {
			if (mail_cache_transaction_flush(ctx, FALSE) < 0) {
				/* If this is a syscall failure, the already
				   flushed changes could still be finished by
				   writing the offsets to .log file. If this is
				   a corruption/lost cache, the offsets will
				   point to a nonexistent file or be ignored.
				   Either way, we don't really need to handle
				   this failure in any special way. */
			}
			/* Regardless of whether the flush succeeded, drop all
			   data that it would have written. This way the flush
			   is attempted only once, but it could still be
			   possible to write new data later. Also don't reset
			   the transaction entirely so that the last partially
			   cached mail can still be accessed from memory. */
			mail_cache_transaction_drop_last_flush(ctx);
		}
	}

	buffer_append(ctx->cache_data, &field_idx, sizeof(field_idx));
	if (fixed_size == UINT_MAX) {
		buffer_append(ctx->cache_data, &data_size32,
			      sizeof(data_size32));
	}

	buffer_append(ctx->cache_data, data, data_size);
	if ((data_size & 3) != 0)
                buffer_append_zero(ctx->cache_data, 4 - (data_size & 3));
}

bool mail_cache_field_want_add(struct mail_cache_transaction_ctx *ctx,
			       uint32_t seq, unsigned int field_idx)
{
	enum mail_cache_decision_type decision;

	mail_cache_transaction_refresh_decisions(ctx);

	decision = mail_cache_field_get_decision(ctx->view->cache, field_idx);
	decision &= ENUM_NEGATE(MAIL_CACHE_DECISION_FORCED);
	switch (decision) {
	case MAIL_CACHE_DECISION_NO:
		return FALSE;
	case MAIL_CACHE_DECISION_TEMP:
		/* add it only if it's newer than what we would drop when
		   purging */
		if (ctx->first_new_seq == 0) {
			ctx->first_new_seq =
				mail_cache_get_first_new_seq(ctx->view->view);
		}
		if (seq < ctx->first_new_seq)
			return FALSE;
		break;
	default:
		break;
	}

	return mail_cache_field_exists(ctx->view, seq, field_idx) == 0;
}

bool mail_cache_field_can_add(struct mail_cache_transaction_ctx *ctx,
			      uint32_t seq, unsigned int field_idx)
{
	enum mail_cache_decision_type decision;

	mail_cache_transaction_refresh_decisions(ctx);

	decision = mail_cache_field_get_decision(ctx->view->cache, field_idx);
	if (decision == (MAIL_CACHE_DECISION_FORCED | MAIL_CACHE_DECISION_NO))
		return FALSE;

	return mail_cache_field_exists(ctx->view, seq, field_idx) == 0;
}

void mail_cache_close_mail(struct mail_cache_transaction_ctx *ctx,
			   uint32_t seq)
{
	if (array_is_created(&ctx->cache_data_wanted_seqs))
		seq_range_array_remove(&ctx->cache_data_wanted_seqs, seq);
}