1 /*
2  * mbsync - mailbox synchronizer
3  * Copyright (C) 2000-2002 Michael R. Elkins <me@mutt.org>
4  * Copyright (C) 2002-2006,2010-2013 Oswald Buddenhagen <ossi@users.sf.net>
5  *
6  *  This program is free software; you can redistribute it and/or modify
7  *  it under the terms of the GNU General Public License as published by
8  *  the Free Software Foundation; either version 2 of the License, or
9  *  (at your option) any later version.
10  *
11  *  This program is distributed in the hope that it will be useful,
12  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
13  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14  *  GNU General Public License for more details.
15  *
16  *  You should have received a copy of the GNU General Public License
17  *  along with this program.  If not, see <http://www.gnu.org/licenses/>.
18  *
19  * As a special exception, mbsync may be linked with the OpenSSL library,
20  * despite that library's more restrictive license.
21  */
22 
23 #include "sync.h"
24 
25 #include <assert.h>
26 #include <stdio.h>
27 #include <limits.h>
28 #include <stdlib.h>
29 #include <stddef.h>
30 #include <unistd.h>
31 #include <time.h>
32 #include <fcntl.h>
33 #include <ctype.h>
34 #include <string.h>
35 #include <errno.h>
36 #include <sys/stat.h>
37 
38 #if !defined(_POSIX_SYNCHRONIZED_IO) || _POSIX_SYNCHRONIZED_IO <= 0
39 # define fdatasync fsync
40 #endif
41 
42 #define JOURNAL_VERSION "4"
43 
44 channel_conf_t global_conf;
45 channel_conf_t *channels;
46 group_conf_t *groups;
47 
48 const char *str_fn[] = { "far side", "near side" }, *str_hl[] = { "push", "pull" };
49 
50 static void ATTR_PRINTFLIKE(1, 2)
debug(const char * msg,...)51 debug( const char *msg, ... )
52 {
53 	va_list va;
54 
55 	va_start( va, msg );
56 	vdebug( DEBUG_SYNC, msg, va );
57 	va_end( va );
58 }
59 
60 static void ATTR_PRINTFLIKE(1, 2)
debugn(const char * msg,...)61 debugn( const char *msg, ... )
62 {
63 	va_list va;
64 
65 	va_start( va, msg );
66 	vdebugn( DEBUG_SYNC, msg, va );
67 	va_end( va );
68 }
69 
70 static void
Fclose(FILE * f,int safe)71 Fclose( FILE *f, int safe )
72 {
73 	if ((safe && (fflush( f ) || (UseFSync && fdatasync( fileno( f ) )))) || fclose( f ) == EOF) {
74 		sys_error( "Error: cannot close file" );
75 		exit( 1 );
76 	}
77 }
78 
79 static void ATTR_PRINTFLIKE(2, 0)
vFprintf(FILE * f,const char * msg,va_list va)80 vFprintf( FILE *f, const char *msg, va_list va )
81 {
82 	int r;
83 
84 	r = vfprintf( f, msg, va );
85 	if (r < 0) {
86 		sys_error( "Error: cannot write file" );
87 		exit( 1 );
88 	}
89 }
90 
91 static void ATTR_PRINTFLIKE(2, 3)
Fprintf(FILE * f,const char * msg,...)92 Fprintf( FILE *f, const char *msg, ... )
93 {
94 	va_list va;
95 
96 	va_start( va, msg );
97 	vFprintf( f, msg, va );
98 	va_end( va );
99 }
100 
101 
102 /* Keep the mailbox driver flag definitions in sync: */
103 /* grep for MAILBOX_DRIVER_FLAG */
104 /* The order is according to alphabetical maildir flag sort */
105 static const char Flags[] = { 'D', 'F', 'P', 'R', 'S', 'T' };
106 
107 static uchar
parse_flags(const char * buf)108 parse_flags( const char *buf )
109 {
110 	uint i, d;
111 	uchar flags;
112 
113 	for (flags = i = d = 0; i < as(Flags); i++)
114 		if (buf[d] == Flags[i]) {
115 			flags |= (1 << i);
116 			d++;
117 		}
118 	return flags;
119 }
120 
121 static uint
make_flags(uchar flags,char * buf)122 make_flags( uchar flags, char *buf )
123 {
124 	uint i, d;
125 
126 	for (i = d = 0; i < as(Flags); i++)
127 		if (flags & (1 << i))
128 			buf[d++] = Flags[i];
129 	buf[d] = 0;
130 	return d;
131 }
132 
133 // This is the (mostly) persistent status of the sync record.
134 // Most of these bits are actually mutually exclusive. It is a
135 // bitfield to allow for easy testing for multiple states.
136 #define S_EXPIRE       (1<<0)  // the entry is being expired (near side message removal scheduled)
137 #define S_EXPIRED      (1<<1)  // the entry is expired (near side message removal confirmed)
138 #define S_PENDING      (1<<2)  // the entry is new and awaits propagation (possibly a retry)
139 #define S_DUMMY(fn)    (1<<(3+(fn)))  // f/n message is only a placeholder
140 #define S_SKIPPED      (1<<5)  // pre-1.4 legacy: the entry was not propagated (message is too big)
141 #define S_DEAD         (1<<7)  // ephemeral: the entry was killed and should be ignored
142 
143 // Ephemeral working set.
144 #define W_NEXPIRE      (1<<0)  // temporary: new expiration state
145 #define W_DELETE       (1<<1)  // ephemeral: flags propagation is a deletion
146 #define W_DEL(fn)      (1<<(2+(fn)))  // ephemeral: f/n message would be subject to expunge
147 #define W_UPGRADE      (1<<4)  // ephemeral: upgrading placeholder, do not apply MaxSize
148 #define W_PURGE        (1<<5)  // ephemeral: placeholder is being nuked
149 
150 typedef struct sync_rec {
151 	struct sync_rec *next;
152 	/* string_list_t *keywords; */
153 	uint uid[2];
154 	message_t *msg[2];
155 	uchar status, wstate, flags, pflags, aflags[2], dflags[2];
156 	char tuid[TUIDL];
157 } sync_rec_t;
158 
159 typedef struct {
160 	int t[2];
161 	void (*cb)( int sts, void *aux ), *aux;
162 	char *dname, *jname, *nname, *lname, *box_name[2];
163 	FILE *jfp, *nfp;
164 	sync_rec_t *srecs, **srecadd;
165 	channel_conf_t *chan;
166 	store_t *ctx[2];
167 	driver_t *drv[2];
168 	const char *orig_name[2];
169 	message_t *msgs[2], *new_msgs[2];
170 	uint_array_alloc_t trashed_msgs[2];
171 	int state[2], lfd, ret, existing, replayed;
172 	uint ref_count, nsrecs, opts[2];
173 	uint new_pending[2], flags_pending[2], trash_pending[2];
174 	uint maxuid[2];     // highest UID that was already propagated
175 	uint oldmaxuid[2];  // highest UID that was already propagated before this run
176 	uint uidval[2];     // UID validity value
177 	uint newuidval[2];  // UID validity obtained from driver
178 	uint finduid[2];    // TUID lookup makes sense only for UIDs >= this
179 	uint maxxfuid;      // highest expired UID on far side
180 	uint oldmaxxfuid;   // highest expired UID on far side before this run
181 	uchar good_flags[2], bad_flags[2];
182 } sync_vars_t;
183 
sync_ref(sync_vars_t * svars)184 static void sync_ref( sync_vars_t *svars ) { ++svars->ref_count; }
185 static void sync_deref( sync_vars_t *svars );
186 static int check_cancel( sync_vars_t *svars );
187 
188 #define AUX &svars->t[t]
189 #define INV_AUX &svars->t[1-t]
190 #define DECL_SVARS \
191 	int t; \
192 	sync_vars_t *svars
193 #define INIT_SVARS(aux) \
194 	t = *(int *)aux; \
195 	svars = (sync_vars_t *)(((char *)(&((int *)aux)[-t])) - offsetof(sync_vars_t, t))
196 #define DECL_INIT_SVARS(aux) \
197 	int t = *(int *)aux; \
198 	sync_vars_t *svars = (sync_vars_t *)(((char *)(&((int *)aux)[-t])) - offsetof(sync_vars_t, t))
199 
200 /* operation dependencies:
201    select(x): -
202    load(x): select(x)
203    new(F), new(N), flags(F), flags(N): load(F) & load(N)
204    find_new(x): new(x)
205    trash(x): flags(x)
206    close(x): trash(x) & find_new(x) & new(!x) // with expunge
207    cleanup: close(F) & close(N)
208 */
209 
210 #define ST_LOADED          (1<<0)
211 #define ST_FIND_OLD        (1<<1)
212 #define ST_SENT_NEW        (1<<2)
213 #define ST_FIND_NEW        (1<<3)
214 #define ST_FOUND_NEW       (1<<4)
215 #define ST_SENT_FLAGS      (1<<5)
216 #define ST_SENT_TRASH      (1<<6)
217 #define ST_CLOSED          (1<<7)
218 #define ST_SENT_CANCEL     (1<<8)
219 #define ST_CANCELED        (1<<9)
220 #define ST_SELECTED        (1<<10)
221 #define ST_DID_EXPUNGE     (1<<11)
222 #define ST_CLOSING         (1<<12)
223 #define ST_CONFIRMED       (1<<13)
224 #define ST_PRESENT         (1<<14)
225 #define ST_SENDING_NEW     (1<<15)
226 
227 
228 static void
create_state(sync_vars_t * svars)229 create_state( sync_vars_t *svars )
230 {
231 	if (!(svars->nfp = fopen( svars->nname, "w" ))) {
232 		sys_error( "Error: cannot create new sync state %s", svars->nname );
233 		exit( 1 );
234 	}
235 }
236 
237 static void ATTR_PRINTFLIKE(2, 3)
jFprintf(sync_vars_t * svars,const char * msg,...)238 jFprintf( sync_vars_t *svars, const char *msg, ... )
239 {
240 	va_list va;
241 
242 	if (JLimit && !--JLimit)
243 		exit( 101 );
244 	if (!svars->jfp) {
245 		create_state( svars );
246 		if (!(svars->jfp = fopen( svars->jname, svars->replayed ? "a" : "w" ))) {
247 			sys_error( "Error: cannot create journal %s", svars->jname );
248 			exit( 1 );
249 		}
250 		setlinebuf( svars->jfp );
251 		if (!svars->replayed)
252 			Fprintf( svars->jfp, JOURNAL_VERSION "\n" );
253 	}
254 	va_start( va, msg );
255 	vFprintf( svars->jfp, msg, va );
256 	va_end( va );
257 	if (JLimit && !--JLimit)
258 		exit( 100 );
259 }
260 
261 #define JLOG_(log_fmt, log_args, dbg_fmt, ...) \
262 	do { \
263 		debug( "-> log: " log_fmt " (" dbg_fmt ")\n", __VA_ARGS__ ); \
264 		jFprintf( svars, log_fmt "\n", deparen(log_args) ); \
265 	} while (0)
266 #define JLOG3(log_fmt, log_args, dbg_fmt) \
267 	JLOG_(log_fmt, log_args, dbg_fmt, deparen(log_args))
268 #define JLOG4(log_fmt, log_args, dbg_fmt, dbg_args) \
269 	JLOG_(log_fmt, log_args, dbg_fmt, deparen(log_args), deparen(dbg_args))
270 #define JLOG_SEL(_1, _2, _3, _4, x, ...) x
271 #define JLOG(...) JLOG_SEL(__VA_ARGS__, JLOG4, JLOG3, NO_JLOG2, NO_JLOG1)(__VA_ARGS__)
272 
273 static void
assign_uid(sync_vars_t * svars,sync_rec_t * srec,int t,uint uid)274 assign_uid( sync_vars_t *svars, sync_rec_t *srec, int t, uint uid )
275 {
276 	srec->uid[t] = uid;
277 	if (uid == svars->maxuid[t] + 1)
278 		svars->maxuid[t] = uid;
279 	srec->status &= ~S_PENDING;
280 	srec->wstate &= ~W_UPGRADE;
281 	srec->tuid[0] = 0;
282 }
283 
284 #define ASSIGN_UID(srec, t, nuid, ...) \
285 	do { \
286 		JLOG( "%c %u %u %u", ("<>"[t], srec->uid[F], srec->uid[N], nuid), __VA_ARGS__ ); \
287 		assign_uid( svars, srec, t, nuid ); \
288 	} while (0)
289 
290 static void
match_tuids(sync_vars_t * svars,int t,message_t * msgs)291 match_tuids( sync_vars_t *svars, int t, message_t *msgs )
292 {
293 	sync_rec_t *srec;
294 	message_t *tmsg, *ntmsg = NULL;
295 	const char *diag;
296 	int num_lost = 0;
297 
298 	for (srec = svars->srecs; srec; srec = srec->next) {
299 		if (srec->status & S_DEAD)
300 			continue;
301 		if (!srec->uid[t] && srec->tuid[0]) {
302 			debug( "pair(%u,%u) TUID %." stringify(TUIDL) "s\n", srec->uid[F], srec->uid[N], srec->tuid );
303 			for (tmsg = ntmsg; tmsg; tmsg = tmsg->next) {
304 				if (tmsg->status & M_DEAD)
305 					continue;
306 				if (tmsg->tuid[0] && !memcmp( tmsg->tuid, srec->tuid, TUIDL )) {
307 					diag = (tmsg == ntmsg) ? "adjacently" : "after gap";
308 					goto mfound;
309 				}
310 			}
311 			for (tmsg = msgs; tmsg != ntmsg; tmsg = tmsg->next) {
312 				if (tmsg->status & M_DEAD)
313 					continue;
314 				if (tmsg->tuid[0] && !memcmp( tmsg->tuid, srec->tuid, TUIDL )) {
315 					diag = "after reset";
316 					goto mfound;
317 				}
318 			}
319 			JLOG( "& %u %u", (srec->uid[F], srec->uid[N]), "TUID lost" );
320 			// Note: status remains S_PENDING.
321 			srec->tuid[0] = 0;
322 			num_lost++;
323 			continue;
324 		  mfound:
325 			tmsg->srec = srec;
326 			srec->msg[t] = tmsg;
327 			ntmsg = tmsg->next;
328 			ASSIGN_UID( srec, t, tmsg->uid, "TUID matched %s", diag );
329 		}
330 	}
331 	if (num_lost)
332 		warn( "Warning: lost track of %d %sed message(s)\n", num_lost, str_hl[t] );
333 }
334 
335 
336 static uchar
sanitize_flags(uchar tflags,sync_vars_t * svars,int t)337 sanitize_flags( uchar tflags, sync_vars_t *svars, int t )
338 {
339 	if (!(DFlags & QUIET)) {
340 		// We complain only once per flag per store - even though _theoretically_
341 		// each mailbox can support different flags according to the IMAP spec.
342 		uchar bflags = tflags & ~(svars->good_flags[t] | svars->bad_flags[t]);
343 		if (bflags) {
344 			char bfbuf[16];
345 			make_flags( bflags, bfbuf );
346 			notice( "Notice: %s store does not support flag(s) '%s'; not propagating.\n", str_fn[t], bfbuf );
347 			svars->bad_flags[t] |= bflags;
348 		}
349 	}
350 	return tflags & svars->good_flags[t];
351 }
352 
353 
354 typedef struct copy_vars {
355 	void (*cb)( int sts, uint uid, struct copy_vars *vars );
356 	void *aux;
357 	sync_rec_t *srec; /* also ->tuid */
358 	message_t *msg;
359 	msg_data_t data;
360 	int minimal;
361 } copy_vars_t;
362 
363 static void msg_fetched( int sts, void *aux );
364 
365 static void
copy_msg(copy_vars_t * vars)366 copy_msg( copy_vars_t *vars )
367 {
368 	DECL_INIT_SVARS(vars->aux);
369 
370 	t ^= 1;
371 	vars->data.flags = vars->msg->flags;
372 	vars->data.date = svars->chan->use_internal_date ? -1 : 0;
373 	svars->drv[t]->fetch_msg( svars->ctx[t], vars->msg, &vars->data, vars->minimal, msg_fetched, vars );
374 }
375 
376 static void msg_stored( int sts, uint uid, void *aux );
377 
378 static void
copy_msg_bytes(char ** out_ptr,const char * in_buf,uint * in_idx,uint in_len,int in_cr,int out_cr)379 copy_msg_bytes( char **out_ptr, const char *in_buf, uint *in_idx, uint in_len, int in_cr, int out_cr )
380 {
381 	char *out = *out_ptr;
382 	uint idx = *in_idx;
383 	if (out_cr != in_cr) {
384 		char c;
385 		if (out_cr) {
386 			for (; idx < in_len; idx++) {
387 				if ((c = in_buf[idx]) != '\r') {
388 					if (c == '\n')
389 						*out++ = '\r';
390 					*out++ = c;
391 				}
392 			}
393 		} else {
394 			for (; idx < in_len; idx++) {
395 				if ((c = in_buf[idx]) != '\r')
396 					*out++ = c;
397 			}
398 		}
399 	} else {
400 		memcpy( out, in_buf + idx, in_len - idx );
401 		out += in_len - idx;
402 		idx = in_len;
403 	}
404 	*out_ptr = out;
405 	*in_idx = idx;
406 }
407 
408 static int
copy_msg_convert(int in_cr,int out_cr,copy_vars_t * vars,int t)409 copy_msg_convert( int in_cr, int out_cr, copy_vars_t *vars, int t )
410 {
411 	char *in_buf = vars->data.data;
412 	uint in_len = vars->data.len;
413 	uint idx = 0, sbreak = 0, ebreak = 0, break2 = UINT_MAX;
414 	uint lines = 0, hdr_crs = 0, bdy_crs = 0, app_cr = 0, extra = 0;
415 	uint add_subj = 0;
416 
417 	if (vars->srec) {
418 	  nloop: ;
419 		uint start = idx;
420 		uint line_crs = 0;
421 		while (idx < in_len) {
422 			char c = in_buf[idx++];
423 			if (c == '\r') {
424 				line_crs++;
425 			} else if (c == '\n') {
426 				if (!ebreak && starts_with_upper( in_buf + start, (int)(in_len - start), "X-TUID: ", 8 )) {
427 					extra = (sbreak = start) - (ebreak = idx);
428 					if (!vars->minimal)
429 						goto oke;
430 				} else {
431 					if (break2 == UINT_MAX && vars->minimal &&
432 					    starts_with_upper( in_buf + start, (int)(in_len - start), "SUBJECT:", 8 )) {
433 						break2 = start + 8;
434 						if (break2 < in_len && in_buf[break2] == ' ')
435 							break2++;
436 					}
437 					lines++;
438 					hdr_crs += line_crs;
439 				}
440 				if (idx - line_crs - 1 == start) {
441 					if (!ebreak)
442 						sbreak = ebreak = start;
443 					if (vars->minimal) {
444 						in_len = idx;
445 						if (break2 == UINT_MAX) {
446 							break2 = start;
447 							add_subj = 1;
448 						}
449 					}
450 					goto oke;
451 				}
452 				goto nloop;
453 			}
454 		}
455 		warn( "Warning: message %u from %s has incomplete header; skipping.\n",
456 		      vars->msg->uid, str_fn[1-t] );
457 		free( in_buf );
458 		return 0;
459 	  oke:
460 		app_cr = out_cr && (!in_cr || hdr_crs);
461 		extra += 8 + TUIDL + app_cr + 1;
462 	}
463 	if (out_cr != in_cr) {
464 		for (; idx < in_len; idx++) {
465 			char c = in_buf[idx];
466 			if (c == '\r')
467 				bdy_crs++;
468 			else if (c == '\n')
469 				lines++;
470 		}
471 		extra -= hdr_crs + bdy_crs;
472 		if (out_cr)
473 			extra += lines;
474 	}
475 
476 	uint dummy_msg_len = 0;
477 	char dummy_msg_buf[180];
478 	static const char dummy_pfx[] = "[placeholder] ";
479 	static const char dummy_subj[] = "Subject: [placeholder] (No Subject)";
480 	static const char dummy_msg[] =
481 		"Having a size of %s, this message is over the MaxSize limit.%s"
482 		"Flag it and sync again (Sync mode ReNew) to fetch its real contents.%s";
483 
484 	if (vars->minimal) {
485 		char sz[32];
486 
487 		if (vars->msg->size < 1024000)
488 			sprintf( sz, "%dKiB", (int)(vars->msg->size >> 10) );
489 		else
490 			sprintf( sz, "%.1fMiB", vars->msg->size / 1048576. );
491 		const char *nl = app_cr ? "\r\n" : "\n";
492 		dummy_msg_len = (uint)sprintf( dummy_msg_buf, dummy_msg, sz, nl, nl );
493 		extra += dummy_msg_len;
494 		extra += add_subj ? strlen(dummy_subj) + app_cr + 1 : strlen(dummy_pfx);
495 	}
496 
497 	vars->data.len = in_len + extra;
498 	if (vars->data.len > INT_MAX) {
499 		warn( "Warning: message %u from %s is too big after conversion; skipping.\n",
500 		      vars->msg->uid, str_fn[1-t] );
501 		free( in_buf );
502 		return 0;
503 	}
504 	char *out_buf = vars->data.data = nfmalloc( vars->data.len );
505 	idx = 0;
506 	if (vars->srec) {
507 		if (break2 < sbreak) {
508 			copy_msg_bytes( &out_buf, in_buf, &idx, break2, in_cr, out_cr );
509 			memcpy( out_buf, dummy_pfx, strlen(dummy_pfx) );
510 			out_buf += strlen(dummy_pfx);
511 		}
512 		copy_msg_bytes( &out_buf, in_buf, &idx, sbreak, in_cr, out_cr );
513 
514 		memcpy( out_buf, "X-TUID: ", 8 );
515 		out_buf += 8;
516 		memcpy( out_buf, vars->srec->tuid, TUIDL );
517 		out_buf += TUIDL;
518 		if (app_cr)
519 			*out_buf++ = '\r';
520 		*out_buf++ = '\n';
521 		idx = ebreak;
522 
523 		if (break2 != UINT_MAX && break2 >= sbreak) {
524 			copy_msg_bytes( &out_buf, in_buf, &idx, break2, in_cr, out_cr );
525 			if (!add_subj) {
526 				memcpy( out_buf, dummy_pfx, strlen(dummy_pfx) );
527 				out_buf += strlen(dummy_pfx);
528 			} else {
529 				memcpy( out_buf, dummy_subj, strlen(dummy_subj) );
530 				out_buf += strlen(dummy_subj);
531 				if (app_cr)
532 					*out_buf++ = '\r';
533 				*out_buf++ = '\n';
534 			}
535 		}
536 	}
537 	copy_msg_bytes( &out_buf, in_buf, &idx, in_len, in_cr, out_cr );
538 
539 	if (vars->minimal)
540 		memcpy( out_buf, dummy_msg_buf, dummy_msg_len );
541 
542 	free( in_buf );
543 	return 1;
544 }
545 
546 static void
msg_fetched(int sts,void * aux)547 msg_fetched( int sts, void *aux )
548 {
549 	copy_vars_t *vars = (copy_vars_t *)aux;
550 	DECL_SVARS;
551 	int scr, tcr;
552 
553 	switch (sts) {
554 	case DRV_OK:
555 		INIT_SVARS(vars->aux);
556 		if (check_cancel( svars )) {
557 			free( vars->data.data );
558 			vars->cb( SYNC_CANCELED, 0, vars );
559 			return;
560 		}
561 
562 		vars->msg->flags = vars->data.flags = sanitize_flags( vars->data.flags, svars, t );
563 
564 		scr = (svars->drv[1-t]->get_caps( svars->ctx[1-t] ) / DRV_CRLF) & 1;
565 		tcr = (svars->drv[t]->get_caps( svars->ctx[t] ) / DRV_CRLF) & 1;
566 		if (vars->srec || scr != tcr) {
567 			if (!copy_msg_convert( scr, tcr, vars, t )) {
568 				vars->cb( SYNC_NOGOOD, 0, vars );
569 				return;
570 			}
571 		}
572 
573 		svars->drv[t]->store_msg( svars->ctx[t], &vars->data, !vars->srec, msg_stored, vars );
574 		break;
575 	case DRV_CANCELED:
576 		vars->cb( SYNC_CANCELED, 0, vars );
577 		break;
578 	case DRV_MSG_BAD:
579 		vars->cb( SYNC_NOGOOD, 0, vars );
580 		break;
581 	default:
582 		vars->cb( SYNC_FAIL, 0, vars );
583 		break;
584 	}
585 }
586 
587 static void
msg_stored(int sts,uint uid,void * aux)588 msg_stored( int sts, uint uid, void *aux )
589 {
590 	copy_vars_t *vars = (copy_vars_t *)aux;
591 	DECL_SVARS;
592 
593 	switch (sts) {
594 	case DRV_OK:
595 		vars->cb( SYNC_OK, uid, vars );
596 		break;
597 	case DRV_CANCELED:
598 		vars->cb( SYNC_CANCELED, 0, vars );
599 		break;
600 	case DRV_MSG_BAD:
601 		INIT_SVARS(vars->aux);
602 		(void)svars;
603 		warn( "Warning: %s refuses to store message %u from %s.\n",
604 		      str_fn[t], vars->msg->uid, str_fn[1-t] );
605 		vars->cb( SYNC_NOGOOD, 0, vars );
606 		break;
607 	default:
608 		vars->cb( SYNC_FAIL, 0, vars );
609 		break;
610 	}
611 }
612 
613 
614 static void sync_bail( sync_vars_t *svars );
615 static void sync_bail2( sync_vars_t *svars );
616 static void sync_bail3( sync_vars_t *svars );
617 static void cancel_done( void *aux );
618 
619 static void
cancel_sync(sync_vars_t * svars)620 cancel_sync( sync_vars_t *svars )
621 {
622 	int t;
623 
624 	for (t = 0; t < 2; t++) {
625 		int other_state = svars->state[1-t];
626 		if (svars->ret & SYNC_BAD(t)) {
627 			cancel_done( AUX );
628 		} else if (!(svars->state[t] & ST_SENT_CANCEL)) {
629 			/* ignore subsequent failures from in-flight commands */
630 			svars->state[t] |= ST_SENT_CANCEL;
631 			svars->drv[t]->cancel_cmds( svars->ctx[t], cancel_done, AUX );
632 		}
633 		if (other_state & ST_CANCELED)
634 			break;
635 	}
636 }
637 
638 static void
cancel_done(void * aux)639 cancel_done( void *aux )
640 {
641 	DECL_INIT_SVARS(aux);
642 
643 	svars->state[t] |= ST_CANCELED;
644 	if (svars->state[1-t] & ST_CANCELED) {
645 		if (svars->nfp) {
646 			Fclose( svars->nfp, 0 );
647 			Fclose( svars->jfp, 0 );
648 		}
649 		sync_bail( svars );
650 	}
651 }
652 
653 static void
store_bad(void * aux)654 store_bad( void *aux )
655 {
656 	DECL_INIT_SVARS(aux);
657 
658 	svars->drv[t]->cancel_store( svars->ctx[t] );
659 	svars->ret |= SYNC_BAD(t);
660 	cancel_sync( svars );
661 }
662 
663 static int
check_cancel(sync_vars_t * svars)664 check_cancel( sync_vars_t *svars )
665 {
666 	return (svars->state[F] | svars->state[N]) & (ST_SENT_CANCEL | ST_CANCELED);
667 }
668 
669 static int
check_ret(int sts,void * aux)670 check_ret( int sts, void *aux )
671 {
672 	DECL_SVARS;
673 
674 	if (sts == DRV_CANCELED)
675 		return 1;
676 	INIT_SVARS(aux);
677 	if (sts == DRV_BOX_BAD) {
678 		svars->ret |= SYNC_FAIL;
679 		cancel_sync( svars );
680 		return 1;
681 	}
682 	return check_cancel( svars );
683 }
684 
685 #define SVARS_CHECK_RET \
686 	DECL_SVARS; \
687 	if (check_ret( sts, aux )) \
688 		return; \
689 	INIT_SVARS(aux)
690 
691 #define SVARS_CHECK_RET_VARS(type) \
692 	type *vars = (type *)aux; \
693 	DECL_SVARS; \
694 	if (check_ret( sts, vars->aux )) { \
695 		free( vars ); \
696 		return; \
697 	} \
698 	INIT_SVARS(vars->aux)
699 
700 #define SVARS_CHECK_CANCEL_RET \
701 	DECL_SVARS; \
702 	if (sts == SYNC_CANCELED) { \
703 		free( vars ); \
704 		return; \
705 	} \
706 	INIT_SVARS(vars->aux)
707 
708 static char *
clean_strdup(const char * s)709 clean_strdup( const char *s )
710 {
711 	char *cs;
712 	uint i;
713 
714 	cs = nfstrdup( s );
715 	for (i = 0; cs[i]; i++)
716 		if (cs[i] == '/')
717 			cs[i] = '!';
718 	return cs;
719 }
720 
721 
722 static sync_rec_t *
upgrade_srec(sync_vars_t * svars,sync_rec_t * srec)723 upgrade_srec( sync_vars_t *svars, sync_rec_t *srec )
724 {
725 	// Create an entry and append it to the current one.
726 	sync_rec_t *nsrec = nfcalloc( sizeof(*nsrec) );
727 	nsrec->next = srec->next;
728 	srec->next = nsrec;
729 	if (svars->srecadd == &srec->next)
730 		svars->srecadd = &nsrec->next;
731 	// Move the placeholder to the new entry.
732 	int t = (srec->status & S_DUMMY(F)) ? F : N;
733 	nsrec->uid[t] = srec->uid[t];
734 	srec->uid[t] = 0;
735 	if (srec->msg[t]) {  // NULL during journal replay; is assigned later.
736 		nsrec->msg[t] = srec->msg[t];
737 		nsrec->msg[t]->srec = nsrec;
738 		srec->msg[t] = NULL;
739 	}
740 	// Mark the original entry for upgrade.
741 	srec->status = (srec->status & ~(S_DUMMY(F)|S_DUMMY(N))) | S_PENDING;
742 	srec->wstate |= W_UPGRADE;
743 	// Mark the placeholder for nuking.
744 	nsrec->wstate = W_PURGE;
745 	nsrec->aflags[t] = F_DELETED;
746 	return nsrec;
747 }
748 
749 static int
prepare_state(sync_vars_t * svars)750 prepare_state( sync_vars_t *svars )
751 {
752 	char *s, *cmname, *csname;
753 	channel_conf_t *chan;
754 
755 	chan = svars->chan;
756 	if (!strcmp( chan->sync_state ? chan->sync_state : global_conf.sync_state, "*" )) {
757 		const char *path = svars->drv[N]->get_box_path( svars->ctx[N] );
758 		if (!path) {
759 			error( "Error: store '%s' does not support in-box sync state\n", chan->stores[N]->name );
760 			return 0;
761 		}
762 		nfasprintf( &svars->dname, "%s/." EXE "state", path );
763 	} else {
764 		csname = clean_strdup( svars->box_name[N] );
765 		if (chan->sync_state)
766 			nfasprintf( &svars->dname, "%s%s", chan->sync_state, csname );
767 		else {
768 			char c = FieldDelimiter;
769 			cmname = clean_strdup( svars->box_name[F] );
770 			nfasprintf( &svars->dname, "%s%c%s%c%s_%c%s%c%s", global_conf.sync_state,
771 			            c, chan->stores[F]->name, c, cmname, c, chan->stores[N]->name, c, csname );
772 			free( cmname );
773 		}
774 		free( csname );
775 		if (!(s = strrchr( svars->dname, '/' ))) {
776 			error( "Error: invalid SyncState location '%s'\n", svars->dname );
777 			return 0;
778 		}
779 		*s = 0;
780 		if (mkdir( svars->dname, 0700 ) && errno != EEXIST) {
781 			sys_error( "Error: cannot create SyncState directory '%s'", svars->dname );
782 			return 0;
783 		}
784 		*s = '/';
785 	}
786 	nfasprintf( &svars->jname, "%s.journal", svars->dname );
787 	nfasprintf( &svars->nname, "%s.new", svars->dname );
788 	nfasprintf( &svars->lname, "%s.lock", svars->dname );
789 	return 1;
790 }
791 
792 static int
lock_state(sync_vars_t * svars)793 lock_state( sync_vars_t *svars )
794 {
795 	struct flock lck;
796 
797 	if (svars->lfd >= 0)
798 		return 1;
799 	memset( &lck, 0, sizeof(lck) );
800 #if SEEK_SET != 0
801 	lck.l_whence = SEEK_SET;
802 #endif
803 #if F_WRLCK != 0
804 	lck.l_type = F_WRLCK;
805 #endif
806 	if ((svars->lfd = open( svars->lname, O_WRONLY|O_CREAT, 0666 )) < 0) {
807 		sys_error( "Error: cannot create lock file %s", svars->lname );
808 		return 0;
809 	}
810 	if (fcntl( svars->lfd, F_SETLK, &lck )) {
811 		error( "Error: channel :%s:%s-:%s:%s is locked\n",
812 		       svars->chan->stores[F]->name, svars->orig_name[F], svars->chan->stores[N]->name, svars->orig_name[N] );
813 		close( svars->lfd );
814 		svars->lfd = -1;
815 		return 0;
816 	}
817 	return 1;
818 }
819 
820 static void
save_state(sync_vars_t * svars)821 save_state( sync_vars_t *svars )
822 {
823 	sync_rec_t *srec;
824 	char fbuf[16]; /* enlarge when support for keywords is added */
825 
826 	// If no change was made, the state is also unmodified.
827 	if (!svars->jfp && !svars->replayed)
828 		return;
829 
830 	if (!svars->nfp)
831 		create_state( svars );
832 	Fprintf( svars->nfp,
833 	         "FarUidValidity %u\nNearUidValidity %u\nMaxPulledUid %u\nMaxPushedUid %u\n",
834 	         svars->uidval[F], svars->uidval[N], svars->maxuid[F], svars->maxuid[N] );
835 	if (svars->maxxfuid)
836 		Fprintf( svars->nfp, "MaxExpiredFarUid %u\n", svars->maxxfuid );
837 	Fprintf( svars->nfp, "\n" );
838 	for (srec = svars->srecs; srec; srec = srec->next) {
839 		if (srec->status & S_DEAD)
840 			continue;
841 		make_flags( srec->flags, fbuf );
842 		Fprintf( svars->nfp, "%u %u %s%s%s\n", srec->uid[F], srec->uid[N],
843 		         (srec->status & S_DUMMY(F)) ? "<" : (srec->status & S_DUMMY(N)) ? ">" : "",
844 		         (srec->status & S_SKIPPED) ? "^" : (srec->status & S_EXPIRED) ? "~" : "", fbuf );
845 	}
846 
847 	Fclose( svars->nfp, 1 );
848 	if (svars->jfp)
849 		Fclose( svars->jfp, 0 );
850 	if (!(DFlags & KEEPJOURNAL)) {
851 		/* order is important! */
852 		if (rename( svars->nname, svars->dname ))
853 			warn( "Warning: cannot commit sync state %s\n", svars->dname );
854 		else if (unlink( svars->jname ))
855 			warn( "Warning: cannot delete journal %s\n", svars->jname );
856 	}
857 }
858 
859 static int
load_state(sync_vars_t * svars)860 load_state( sync_vars_t *svars )
861 {
862 	sync_rec_t *srec, *nsrec;
863 	char *s;
864 	FILE *jfp;
865 	uint ll;
866 	uint maxxnuid = 0;
867 	char c;
868 	struct stat st;
869 	char fbuf[16]; /* enlarge when support for keywords is added */
870 	char buf[128], buf1[64], buf2[64];
871 
872 	if ((jfp = fopen( svars->dname, "r" ))) {
873 		if (!lock_state( svars ))
874 			goto jbail;
875 		debug( "reading sync state %s ...\n", svars->dname );
876 		int line = 0;
877 		while (fgets( buf, sizeof(buf), jfp )) {
878 			line++;
879 			if (!(ll = strlen( buf )) || buf[ll - 1] != '\n') {
880 				error( "Error: incomplete sync state header entry at %s:%d\n", svars->dname, line );
881 			  jbail:
882 				fclose( jfp );
883 				return 0;
884 			}
885 			if (ll == 1)
886 				goto gothdr;
887 			if (line == 1 && isdigit( buf[0] )) {  // Pre-1.1 legacy
888 				if (sscanf( buf, "%63s %63s", buf1, buf2 ) != 2 ||
889 				    sscanf( buf1, "%u:%u", &svars->uidval[F], &svars->maxuid[F] ) < 2 ||
890 				    sscanf( buf2, "%u:%u:%u", &svars->uidval[N], &maxxnuid, &svars->maxuid[N] ) < 3) {
891 					error( "Error: invalid sync state header in %s\n", svars->dname );
892 					goto jbail;
893 				}
894 				goto gothdr;
895 			}
896 			uint uid;
897 			if (sscanf( buf, "%63s %u", buf1, &uid ) != 2) {
898 				error( "Error: malformed sync state header entry at %s:%d\n", svars->dname, line );
899 				goto jbail;
900 			}
901 			if (!strcmp( buf1, "FarUidValidity" ) || !strcmp( buf1, "MasterUidValidity" ) /* Pre-1.4 legacy */)
902 				svars->uidval[F] = uid;
903 			else if (!strcmp( buf1, "NearUidValidity" ) || !strcmp( buf1, "SlaveUidValidity" ) /* Pre-1.4 legacy */)
904 				svars->uidval[N] = uid;
905 			else if (!strcmp( buf1, "MaxPulledUid" ))
906 				svars->maxuid[F] = uid;
907 			else if (!strcmp( buf1, "MaxPushedUid" ))
908 				svars->maxuid[N] = uid;
909 			else if (!strcmp( buf1, "MaxExpiredFarUid" ) || !strcmp( buf1, "MaxExpiredMasterUid" ) /* Pre-1.4 legacy */)
910 				svars->maxxfuid = uid;
911 			else if (!strcmp( buf1, "MaxExpiredSlaveUid" ))  // Pre-1.3 legacy
912 				maxxnuid = uid;
913 			else {
914 				error( "Error: unrecognized sync state header entry at %s:%d\n", svars->dname, line );
915 				goto jbail;
916 			}
917 		}
918 		error( "Error: unterminated sync state header in %s\n", svars->dname );
919 		goto jbail;
920 	  gothdr:
921 		while (fgets( buf, sizeof(buf), jfp )) {
922 			line++;
923 			if (!(ll = strlen( buf )) || buf[--ll] != '\n') {
924 				error( "Error: incomplete sync state entry at %s:%d\n", svars->dname, line );
925 				goto jbail;
926 			}
927 			buf[ll] = 0;
928 			fbuf[0] = 0;
929 			uint t1, t2;
930 			if (sscanf( buf, "%u %u %15s", &t1, &t2, fbuf ) < 2) {
931 				error( "Error: invalid sync state entry at %s:%d\n", svars->dname, line );
932 				goto jbail;
933 			}
934 			srec = nfcalloc( sizeof(*srec) );
935 			srec->uid[F] = t1;
936 			srec->uid[N] = t2;
937 			s = fbuf;
938 			if (*s == '<') {
939 				s++;
940 				srec->status = S_DUMMY(F);
941 			} else if (*s == '>') {
942 				s++;
943 				srec->status = S_DUMMY(N);
944 			}
945 			if (*s == '^') {  // Pre-1.4 legacy
946 				s++;
947 				srec->status = S_SKIPPED;
948 			} else if (*s == '~' || *s == 'X' /* Pre-1.3 legacy */) {
949 				s++;
950 				srec->status = S_EXPIRE | S_EXPIRED;
951 			} else if (srec->uid[F] == (uint)-1) {  // Pre-1.3 legacy
952 				srec->uid[F] = 0;
953 				srec->status = S_SKIPPED;
954 			} else if (srec->uid[N] == (uint)-1) {
955 				srec->uid[N] = 0;
956 				srec->status = S_SKIPPED;
957 			}
958 			srec->flags = parse_flags( s );
959 			debug( "  entry (%u,%u,%u,%s%s)\n", srec->uid[F], srec->uid[N], srec->flags,
960 			       (srec->status & S_SKIPPED) ? "SKIP" : (srec->status & S_EXPIRED) ? "XPIRE" : "",
961 			       (srec->status & S_DUMMY(F)) ? ",F-DUMMY" : (srec->status & S_DUMMY(N)) ? ",N-DUMMY" : "" );
962 			*svars->srecadd = srec;
963 			svars->srecadd = &srec->next;
964 			svars->nsrecs++;
965 		}
966 		fclose( jfp );
967 		svars->existing = 1;
968 	} else {
969 		if (errno != ENOENT) {
970 			sys_error( "Error: cannot read sync state %s", svars->dname );
971 			return 0;
972 		}
973 		svars->existing = 0;
974 	}
975 
976 	// This is legacy support for pre-1.3 sync states.
977 	if (maxxnuid) {
978 		uint minwuid = UINT_MAX;
979 		for (srec = svars->srecs; srec; srec = srec->next) {
980 			if ((srec->status & (S_DEAD | S_SKIPPED | S_PENDING)) || !srec->uid[F])
981 				continue;
982 			if (srec->status & S_EXPIRED) {
983 				if (!srec->uid[N]) {
984 					// The expired message was already gone.
985 					continue;
986 				}
987 				// The expired message was not expunged yet, so re-examine it.
988 				// This will happen en masse, so just extend the bulk fetch.
989 			} else {
990 				if (srec->uid[N] && maxxnuid >= srec->uid[N]) {
991 					// The non-expired message is in the generally expired range,
992 					// so don't make it contribute to the bulk fetch.
993 					continue;
994 				}
995 				// Usual non-expired message.
996 			}
997 			if (minwuid > srec->uid[F])
998 				minwuid = srec->uid[F];
999 		}
1000 		svars->maxxfuid = minwuid - 1;
1001 	}
1002 
1003 	int line = 0;
1004 	if ((jfp = fopen( svars->jname, "r" ))) {
1005 		if (!lock_state( svars ))
1006 			goto jbail;
1007 		if (!stat( svars->nname, &st ) && fgets( buf, sizeof(buf), jfp )) {
1008 			debug( "recovering journal ...\n" );
1009 			if (!(ll = strlen( buf )) || buf[--ll] != '\n') {
1010 				error( "Error: incomplete journal header in %s\n", svars->jname );
1011 				goto jbail;
1012 			}
1013 			buf[ll] = 0;
1014 			if (!equals( buf, (int)ll, JOURNAL_VERSION, strlen(JOURNAL_VERSION) )) {
1015 				error( "Error: incompatible journal version "
1016 				                 "(got %s, expected " JOURNAL_VERSION ")\n", buf );
1017 				goto jbail;
1018 			}
1019 			srec = NULL;
1020 			line = 1;
1021 			while (fgets( buf, sizeof(buf), jfp )) {
1022 				line++;
1023 				if (!(ll = strlen( buf )) || buf[--ll] != '\n') {
1024 					error( "Error: incomplete journal entry at %s:%d\n", svars->jname, line );
1025 					goto jbail;
1026 				}
1027 				buf[ll] = 0;
1028 				int tn;
1029 				uint t1, t2, t3, t4;
1030 				if ((c = buf[0]) == '#' ?
1031 				      (tn = 0, (sscanf( buf + 2, "%u %u %n", &t1, &t2, &tn ) < 2) || !tn || (ll - (uint)tn != TUIDL + 2)) :
1032 				      c == '!' ?
1033 				        (sscanf( buf + 2, "%u", &t1 ) != 1) :
1034 				        c == 'N' || c == 'F' || c == 'T' || c == '+' || c == '&' || c == '-' || c == '=' || c == '_' || c == '|' ?
1035 				          (sscanf( buf + 2, "%u %u", &t1, &t2 ) != 2) :
1036 				          c != '^' ?
1037 				            (sscanf( buf + 2, "%u %u %u", &t1, &t2, &t3 ) != 3) :
1038 				            (sscanf( buf + 2, "%u %u %u %u", &t1, &t2, &t3, &t4 ) != 4))
1039 				{
1040 					error( "Error: malformed journal entry at %s:%d\n", svars->jname, line );
1041 					goto jbail;
1042 				}
1043 				if (c == 'N')
1044 					svars->maxuid[t1] = t2;
1045 				else if (c == 'F')
1046 					svars->finduid[t1] = t2;
1047 				else if (c == 'T')
1048 					*uint_array_append( &svars->trashed_msgs[t1] ) = t2;
1049 				else if (c == '!')
1050 					svars->maxxfuid = t1;
1051 				else if (c == '|') {
1052 					svars->uidval[F] = t1;
1053 					svars->uidval[N] = t2;
1054 				} else if (c == '+') {
1055 					srec = nfcalloc( sizeof(*srec) );
1056 					srec->uid[F] = t1;
1057 					srec->uid[N] = t2;
1058 					debug( "  new entry(%u,%u)\n", t1, t2 );
1059 					srec->status = S_PENDING;
1060 					*svars->srecadd = srec;
1061 					svars->srecadd = &srec->next;
1062 					svars->nsrecs++;
1063 				} else {
1064 					for (nsrec = srec; srec; srec = srec->next)
1065 						if (srec->uid[F] == t1 && srec->uid[N] == t2)
1066 							goto syncfnd;
1067 					for (srec = svars->srecs; srec != nsrec; srec = srec->next)
1068 						if (srec->uid[F] == t1 && srec->uid[N] == t2)
1069 							goto syncfnd;
1070 					error( "Error: journal entry at %s:%d refers to non-existing sync state entry\n", svars->jname, line );
1071 					goto jbail;
1072 				  syncfnd:
1073 					debugn( "  entry(%u,%u,%u) ", srec->uid[F], srec->uid[N], srec->flags );
1074 					switch (c) {
1075 					case '-':
1076 						debug( "killed\n" );
1077 						srec->status = S_DEAD;
1078 						break;
1079 					case '=':
1080 						debug( "aborted\n" );
1081 						if (svars->maxxfuid < srec->uid[F])
1082 							svars->maxxfuid = srec->uid[F];
1083 						srec->status = S_DEAD;
1084 						break;
1085 					case '#':
1086 						memcpy( srec->tuid, buf + tn + 2, TUIDL );
1087 						debug( "TUID now %." stringify(TUIDL) "s\n", srec->tuid );
1088 						break;
1089 					case '&':
1090 						debug( "TUID %." stringify(TUIDL) "s lost\n", srec->tuid );
1091 						srec->tuid[0] = 0;
1092 						break;
1093 					case '<':
1094 						debug( "far side now %u\n", t3 );
1095 						assign_uid( svars, srec, F, t3 );
1096 						break;
1097 					case '>':
1098 						debug( "near side now %u\n", t3 );
1099 						assign_uid( svars, srec, N, t3 );
1100 						break;
1101 					case '*':
1102 						debug( "flags now %u\n", t3 );
1103 						srec->flags = (uchar)t3;
1104 						srec->aflags[F] = srec->aflags[N] = 0;
1105 						srec->wstate &= ~W_PURGE;
1106 						break;
1107 					case '~':
1108 						debug( "status now %#x\n", t3 );
1109 						srec->status = (uchar)t3;
1110 						break;
1111 					case '_':
1112 						debug( "has placeholder now\n" );
1113 						srec->status = S_PENDING;  // Pre-1.4 legacy only
1114 						srec->status |= !srec->uid[F] ? S_DUMMY(F) : S_DUMMY(N);
1115 						break;
1116 					case '^':
1117 						debug( "is being upgraded, flags %u, srec flags %u\n", t3, t4 );
1118 						srec->pflags = (uchar)t3;
1119 						srec->flags = (uchar)t4;
1120 						srec = upgrade_srec( svars, srec );
1121 						break;
1122 					default:
1123 						error( "Error: unrecognized journal entry at %s:%d\n", svars->jname, line );
1124 						goto jbail;
1125 					}
1126 				}
1127 			}
1128 		}
1129 		fclose( jfp );
1130 	} else {
1131 		if (errno != ENOENT) {
1132 			sys_error( "Error: cannot read journal %s", svars->jname );
1133 			return 0;
1134 		}
1135 	}
1136 	svars->replayed = line;
1137 
1138 	return 1;
1139 }
1140 
1141 static void
delete_state(sync_vars_t * svars)1142 delete_state( sync_vars_t *svars )
1143 {
1144 	unlink( svars->nname );
1145 	unlink( svars->jname );
1146 	if (unlink( svars->dname ) || unlink( svars->lname )) {
1147 		sys_error( "Error: channel %s: sync state cannot be deleted", svars->chan->name );
1148 		svars->ret = SYNC_FAIL;
1149 	}
1150 }
1151 
1152 static void box_confirmed( int sts, uint uidvalidity, void *aux );
1153 static void box_confirmed2( sync_vars_t *svars, int t );
1154 static void box_deleted( int sts, void *aux );
1155 static void box_created( int sts, void *aux );
1156 static void box_opened( int sts, uint uidvalidity, void *aux );
1157 static void box_opened2( sync_vars_t *svars, int t );
1158 static void load_box( sync_vars_t *svars, int t, uint minwuid, uint_array_t mexcs );
1159 
1160 void
sync_boxes(store_t * ctx[],const char * const names[],int present[],channel_conf_t * chan,void (* cb)(int sts,void * aux),void * aux)1161 sync_boxes( store_t *ctx[], const char * const names[], int present[], channel_conf_t *chan,
1162             void (*cb)( int sts, void *aux ), void *aux )
1163 {
1164 	sync_vars_t *svars;
1165 	int t;
1166 
1167 	svars = nfcalloc( sizeof(*svars) );
1168 	svars->t[1] = 1;
1169 	svars->ref_count = 1;
1170 	svars->cb = cb;
1171 	svars->aux = aux;
1172 	svars->ctx[0] = ctx[0];
1173 	svars->ctx[1] = ctx[1];
1174 	svars->chan = chan;
1175 	svars->lfd = -1;
1176 	svars->uidval[0] = svars->uidval[1] = UIDVAL_BAD;
1177 	svars->srecadd = &svars->srecs;
1178 
1179 	for (t = 0; t < 2; t++) {
1180 		svars->orig_name[t] =
1181 			(!names[t] || (ctx[t]->conf->map_inbox && !strcmp( ctx[t]->conf->map_inbox, names[t] ))) ?
1182 				"INBOX" : names[t];
1183 		if (!ctx[t]->conf->flat_delim[0]) {
1184 			svars->box_name[t] = nfstrdup( svars->orig_name[t] );
1185 		} else if (map_name( svars->orig_name[t], &svars->box_name[t], 0, "/", ctx[t]->conf->flat_delim ) < 0) {
1186 			error( "Error: canonical mailbox name '%s' contains flattened hierarchy delimiter\n", svars->orig_name[t] );
1187 		  bail3:
1188 			svars->ret = SYNC_FAIL;
1189 			sync_bail3( svars );
1190 			return;
1191 		}
1192 		svars->drv[t] = ctx[t]->driver;
1193 		svars->drv[t]->set_bad_callback( ctx[t], store_bad, AUX );
1194 	}
1195 	/* Both boxes must be fully set up at this point, so that error exit paths
1196 	 * don't run into uninitialized variables. */
1197 	for (t = 0; t < 2; t++) {
1198 		switch (svars->drv[t]->select_box( ctx[t], svars->box_name[t] )) {
1199 		case DRV_STORE_BAD:
1200 			store_bad( AUX );
1201 			return;
1202 		case DRV_BOX_BAD:
1203 			goto bail3;
1204 		}
1205 	}
1206 
1207 	if (!prepare_state( svars )) {
1208 		svars->ret = SYNC_FAIL;
1209 		sync_bail2( svars );
1210 		return;
1211 	}
1212 	if (!load_state( svars )) {
1213 		svars->ret = SYNC_FAIL;
1214 		sync_bail( svars );
1215 		return;
1216 	}
1217 
1218 	sync_ref( svars );
1219 	for (t = 0; ; t++) {
1220 		info( "Opening %s box %s...\n", str_fn[t], svars->orig_name[t] );
1221 		if (present[t] == BOX_ABSENT)
1222 			box_confirmed2( svars, t );
1223 		else
1224 			svars->drv[t]->open_box( ctx[t], box_confirmed, AUX );
1225 		if (t || check_cancel( svars ))
1226 			break;
1227 	}
1228 	sync_deref( svars );
1229 }
1230 
1231 static void
box_confirmed(int sts,uint uidvalidity,void * aux)1232 box_confirmed( int sts, uint uidvalidity, void *aux )
1233 {
1234 	DECL_SVARS;
1235 
1236 	if (sts == DRV_CANCELED)
1237 		return;
1238 	INIT_SVARS(aux);
1239 	if (check_cancel( svars ))
1240 		return;
1241 
1242 	if (sts == DRV_OK) {
1243 		svars->state[t] |= ST_PRESENT;
1244 		svars->newuidval[t] = uidvalidity;
1245 	}
1246 	box_confirmed2( svars, t );
1247 }
1248 
1249 static void
box_confirmed2(sync_vars_t * svars,int t)1250 box_confirmed2( sync_vars_t *svars, int t )
1251 {
1252 	svars->state[t] |= ST_CONFIRMED;
1253 	if (!(svars->state[1-t] & ST_CONFIRMED))
1254 		return;
1255 
1256 	sync_ref( svars );
1257 	for (t = 0; ; t++) {
1258 		if (!(svars->state[t] & ST_PRESENT)) {
1259 			if (!(svars->state[1-t] & ST_PRESENT)) {
1260 				if (!svars->existing) {
1261 					error( "Error: channel %s: both far side %s and near side %s cannot be opened.\n",
1262 					       svars->chan->name, svars->orig_name[F], svars->orig_name[N] );
1263 				  bail:
1264 					svars->ret = SYNC_FAIL;
1265 				} else {
1266 					/* This can legitimately happen if a deletion propagation was interrupted.
1267 					 * We have no place to record this transaction, so we just assume it.
1268 					 * Of course this bears the danger of clearing the state if both mailboxes
1269 					 * temorarily cannot be opened for some weird reason (while the stores can). */
1270 					delete_state( svars );
1271 				}
1272 			  done:
1273 				sync_bail( svars );
1274 				break;
1275 			}
1276 			if (svars->existing) {
1277 				if (!(svars->chan->ops[1-t] & OP_REMOVE)) {
1278 					error( "Error: channel %s: %s box %s cannot be opened.\n",
1279 					       svars->chan->name, str_fn[t], svars->orig_name[t] );
1280 					goto bail;
1281 				}
1282 				if (svars->drv[1-t]->confirm_box_empty( svars->ctx[1-t] ) != DRV_OK) {
1283 					warn( "Warning: channel %s: %s box %s cannot be opened and %s box %s is not empty.\n",
1284 					      svars->chan->name, str_fn[t], svars->orig_name[t], str_fn[1-t], svars->orig_name[1-t] );
1285 					goto done;
1286 				}
1287 				info( "Deleting %s box %s...\n", str_fn[1-t], svars->orig_name[1-t] );
1288 				svars->drv[1-t]->delete_box( svars->ctx[1-t], box_deleted, INV_AUX );
1289 			} else {
1290 				if (!(svars->chan->ops[t] & OP_CREATE)) {
1291 					box_opened( DRV_BOX_BAD, UIDVAL_BAD, AUX );
1292 				} else {
1293 					info( "Creating %s box %s...\n", str_fn[t], svars->orig_name[t] );
1294 					svars->drv[t]->create_box( svars->ctx[t], box_created, AUX );
1295 				}
1296 			}
1297 		} else {
1298 			box_opened2( svars, t );
1299 		}
1300 		if (t || check_cancel( svars ))
1301 			break;
1302 	}
1303 	sync_deref( svars );
1304 }
1305 
1306 static void
box_deleted(int sts,void * aux)1307 box_deleted( int sts, void *aux )
1308 {
1309 	DECL_SVARS;
1310 
1311 	if (check_ret( sts, aux ))
1312 		return;
1313 	INIT_SVARS(aux);
1314 
1315 	delete_state( svars );
1316 	svars->drv[t]->finish_delete_box( svars->ctx[t] );
1317 	sync_bail( svars );
1318 }
1319 
1320 static void
box_created(int sts,void * aux)1321 box_created( int sts, void *aux )
1322 {
1323 	DECL_SVARS;
1324 
1325 	if (check_ret( sts, aux ))
1326 		return;
1327 	INIT_SVARS(aux);
1328 
1329 	svars->drv[t]->open_box( svars->ctx[t], box_opened, AUX );
1330 }
1331 
1332 static void
box_opened(int sts,uint uidvalidity,void * aux)1333 box_opened( int sts, uint uidvalidity, void *aux )
1334 {
1335 	DECL_SVARS;
1336 
1337 	if (sts == DRV_CANCELED)
1338 		return;
1339 	INIT_SVARS(aux);
1340 	if (check_cancel( svars ))
1341 		return;
1342 
1343 	if (sts == DRV_BOX_BAD) {
1344 		error( "Error: channel %s: %s box %s cannot be opened.\n",
1345 		       svars->chan->name, str_fn[t], svars->orig_name[t] );
1346 		svars->ret = SYNC_FAIL;
1347 		sync_bail( svars );
1348 	} else {
1349 		svars->newuidval[t] = uidvalidity;
1350 		box_opened2( svars, t );
1351 	}
1352 }
1353 
1354 static void
box_opened2(sync_vars_t * svars,int t)1355 box_opened2( sync_vars_t *svars, int t )
1356 {
1357 	store_t *ctx[2];
1358 	channel_conf_t *chan;
1359 	sync_rec_t *srec;
1360 	uint_array_alloc_t mexcs;
1361 	uint opts[2], fails, minwuid;
1362 
1363 	svars->state[t] |= ST_SELECTED;
1364 	if (!(svars->state[1-t] & ST_SELECTED))
1365 		return;
1366 	ctx[0] = svars->ctx[0];
1367 	ctx[1] = svars->ctx[1];
1368 	chan = svars->chan;
1369 
1370 	fails = 0;
1371 	for (t = 0; t < 2; t++)
1372 		if (svars->uidval[t] != UIDVAL_BAD && svars->uidval[t] != svars->newuidval[t])
1373 			fails++;
1374 	// If only one side changed UIDVALIDITY, we will try to re-approve it further down.
1375 	if (fails == 2) {
1376 		error( "Error: channel %s: UIDVALIDITY of both far side %s and near side %s changed.\n",
1377 		       svars->chan->name, svars->orig_name[F], svars->orig_name[N]);
1378 	  bail:
1379 		svars->ret = SYNC_FAIL;
1380 		sync_bail( svars );
1381 		return;
1382 	}
1383 
1384 	if (!lock_state( svars ))
1385 		goto bail;
1386 
1387 	opts[F] = opts[N] = 0;
1388 	if (fails)
1389 		opts[F] = opts[N] = OPEN_OLD|OPEN_OLD_IDS;
1390 	for (t = 0; t < 2; t++) {
1391 		if (chan->ops[t] & (OP_DELETE|OP_FLAGS)) {
1392 			opts[t] |= OPEN_SETFLAGS;
1393 			opts[1-t] |= OPEN_OLD;
1394 			if (chan->ops[t] & OP_FLAGS)
1395 				opts[1-t] |= OPEN_FLAGS;
1396 		}
1397 		if (chan->ops[t] & (OP_NEW|OP_RENEW)) {
1398 			opts[t] |= OPEN_APPEND;
1399 			if (chan->ops[t] & OP_NEW) {
1400 				opts[1-t] |= OPEN_NEW;
1401 				if (chan->stores[t]->max_size != UINT_MAX)
1402 					opts[1-t] |= OPEN_FLAGS|OPEN_NEW_SIZE;
1403 			}
1404 			if (chan->ops[t] & OP_RENEW) {
1405 				opts[t] |= OPEN_OLD|OPEN_FLAGS|OPEN_SETFLAGS;
1406 				opts[1-t] |= OPEN_OLD|OPEN_FLAGS;
1407 			}
1408 			if (chan->ops[t] & OP_EXPUNGE)  // Don't propagate doomed msgs
1409 				opts[1-t] |= OPEN_FLAGS;
1410 		}
1411 		if (chan->ops[t] & OP_EXPUNGE) {
1412 			opts[t] |= OPEN_EXPUNGE;
1413 			if (chan->stores[t]->trash) {
1414 				if (!chan->stores[t]->trash_only_new)
1415 					opts[t] |= OPEN_OLD;
1416 				opts[t] |= OPEN_NEW|OPEN_FLAGS;
1417 			} else if (chan->stores[1-t]->trash && chan->stores[1-t]->trash_remote_new)
1418 				opts[t] |= OPEN_NEW|OPEN_FLAGS;
1419 		}
1420 	}
1421 	if ((chan->ops[N] & (OP_NEW|OP_RENEW|OP_FLAGS)) && chan->max_messages)
1422 		opts[N] |= OPEN_OLD|OPEN_NEW|OPEN_FLAGS;
1423 	if (svars->replayed)
1424 		for (srec = svars->srecs; srec; srec = srec->next) {
1425 			if (srec->status & S_DEAD)
1426 				continue;
1427 			if (srec->tuid[0]) {
1428 				if (!srec->uid[F])
1429 					opts[F] |= OPEN_NEW|OPEN_FIND, svars->state[F] |= ST_FIND_OLD;
1430 				else if (!srec->uid[N])
1431 					opts[N] |= OPEN_NEW|OPEN_FIND, svars->state[N] |= ST_FIND_OLD;
1432 				else
1433 					warn( "Warning: sync record (%u,%u) has stray TUID. Ignoring.\n", srec->uid[F], srec->uid[N] );
1434 			}
1435 			if (srec->wstate & W_PURGE) {
1436 				t = srec->uid[F] ? F : N;
1437 				opts[t] |= OPEN_SETFLAGS;
1438 			}
1439 			if (srec->wstate & W_UPGRADE) {
1440 				t = !srec->uid[F] ? F : N;
1441 				opts[t] |= OPEN_APPEND;
1442 				opts[1-t] |= OPEN_OLD;
1443 			}
1444 		}
1445 	svars->opts[F] = svars->drv[F]->prepare_load_box( ctx[F], opts[F] );
1446 	svars->opts[N] = svars->drv[N]->prepare_load_box( ctx[N], opts[N] );
1447 
1448 	ARRAY_INIT( &mexcs );
1449 	if (svars->opts[F] & OPEN_OLD) {
1450 		if (chan->max_messages) {
1451 			/* When messages have been expired on the near side, the far side fetch is split into
1452 			 * two ranges: The bulk fetch which corresponds with the most recent messages, and an
1453 			 * exception list of messages which would have been expired if they weren't important. */
1454 			debug( "preparing far side selection - max expired far uid is %u\n", svars->maxxfuid );
1455 			/* First, find out the lower bound for the bulk fetch. */
1456 			minwuid = svars->maxxfuid + 1;
1457 			/* Next, calculate the exception fetch. */
1458 			for (srec = svars->srecs; srec; srec = srec->next) {
1459 				if (srec->status & S_DEAD)
1460 					continue;
1461 				if (!srec->uid[F])
1462 					continue;  // No message; other state is irrelevant
1463 				if (srec->uid[F] >= minwuid)
1464 					continue;  // Message is in non-expired range
1465 				if ((svars->opts[F] & OPEN_NEW) && srec->uid[F] >= svars->maxuid[F])
1466 					continue;  // Message is in expired range, but new range overlaps that
1467 				if (!srec->uid[N] && !(srec->status & S_PENDING))
1468 					continue;  // Only actually paired up messages matter
1469 				// The pair is alive, but outside the bulk range
1470 				*uint_array_append( &mexcs ) = srec->uid[F];
1471 			}
1472 			sort_uint_array( mexcs.array );
1473 		} else {
1474 			minwuid = 1;
1475 		}
1476 	} else {
1477 		minwuid = UINT_MAX;
1478 	}
1479 	sync_ref( svars );
1480 	load_box( svars, F, minwuid, mexcs.array );
1481 	if (!check_cancel( svars ))
1482 		load_box( svars, N, (svars->opts[N] & OPEN_OLD) ? 1 : UINT_MAX, (uint_array_t){ NULL, 0 } );
1483 	sync_deref( svars );
1484 }
1485 
1486 static uint
get_seenuid(sync_vars_t * svars,int t)1487 get_seenuid( sync_vars_t *svars, int t )
1488 {
1489 	uint seenuid = 0;
1490 	for (sync_rec_t *srec = svars->srecs; srec; srec = srec->next)
1491 		if (!(srec->status & S_DEAD) && seenuid < srec->uid[t])
1492 			seenuid = srec->uid[t];
1493 	return seenuid;
1494 }
1495 
1496 static void box_loaded( int sts, message_t *msgs, int total_msgs, int recent_msgs, void *aux );
1497 
1498 static void
load_box(sync_vars_t * svars,int t,uint minwuid,uint_array_t mexcs)1499 load_box( sync_vars_t *svars, int t, uint minwuid, uint_array_t mexcs )
1500 {
1501 	uint maxwuid = 0, pairuid = UINT_MAX;
1502 
1503 	if (svars->opts[t] & OPEN_NEW) {
1504 		if (minwuid > svars->maxuid[t] + 1)
1505 			minwuid = svars->maxuid[t] + 1;
1506 		maxwuid = UINT_MAX;
1507 		if (svars->opts[t] & OPEN_OLD_IDS)  // Implies OPEN_OLD
1508 			pairuid = get_seenuid( svars, t );
1509 	} else if (svars->opts[t] & OPEN_OLD) {
1510 		maxwuid = get_seenuid( svars, t );
1511 	}
1512 	info( "Loading %s box...\n", str_fn[t] );
1513 	svars->drv[t]->load_box( svars->ctx[t], minwuid, maxwuid, svars->finduid[t], pairuid, svars->maxuid[t], mexcs, box_loaded, AUX );
1514 }
1515 
1516 typedef struct {
1517 	void *aux;
1518 	sync_rec_t *srec;
1519 	int aflags, dflags;
1520 } flag_vars_t;
1521 
1522 typedef struct {
1523 	uint uid;
1524 	sync_rec_t *srec;
1525 } sync_rec_map_t;
1526 
1527 static void flags_set( int sts, void *aux );
1528 static void flags_set_p2( sync_vars_t *svars, sync_rec_t *srec, int t );
1529 static void msgs_flags_set( sync_vars_t *svars, int t );
1530 static void msg_copied( int sts, uint uid, copy_vars_t *vars );
1531 static void msgs_copied( sync_vars_t *svars, int t );
1532 
1533 static void
box_loaded(int sts,message_t * msgs,int total_msgs,int recent_msgs,void * aux)1534 box_loaded( int sts, message_t *msgs, int total_msgs, int recent_msgs, void *aux )
1535 {
1536 	DECL_SVARS;
1537 	sync_rec_t *srec;
1538 	sync_rec_map_t *srecmap;
1539 	message_t *tmsg;
1540 	flag_vars_t *fv;
1541 	int no[2], del[2], alive, todel;
1542 	uchar sflags, nflags, aflags, dflags;
1543 	uint hashsz, idx;
1544 
1545 	if (check_ret( sts, aux ))
1546 		return;
1547 	INIT_SVARS(aux);
1548 	svars->state[t] |= ST_LOADED;
1549 	svars->msgs[t] = msgs;
1550 	info( "%s: %d messages, %d recent\n", str_fn[t], total_msgs, recent_msgs );
1551 
1552 	if (svars->state[t] & ST_FIND_OLD) {
1553 		debug( "matching previously copied messages on %s\n", str_fn[t] );
1554 		match_tuids( svars, t, msgs );
1555 	}
1556 
1557 	debug( "matching messages on %s against sync records\n", str_fn[t] );
1558 	hashsz = bucketsForSize( svars->nsrecs * 3 );
1559 	srecmap = nfcalloc( hashsz * sizeof(*srecmap) );
1560 	for (srec = svars->srecs; srec; srec = srec->next) {
1561 		if (srec->status & S_DEAD)
1562 			continue;
1563 		uint uid = srec->uid[t];
1564 		if (!uid)
1565 			continue;
1566 		idx = (uint)(uid * 1103515245U) % hashsz;
1567 		while (srecmap[idx].uid)
1568 			if (++idx == hashsz)
1569 				idx = 0;
1570 		srecmap[idx].uid = uid;
1571 		srecmap[idx].srec = srec;
1572 	}
1573 	for (tmsg = svars->msgs[t]; tmsg; tmsg = tmsg->next) {
1574 		if (tmsg->srec) /* found by TUID */
1575 			continue;
1576 		uint uid = tmsg->uid;
1577 		idx = (uint)(uid * 1103515245U) % hashsz;
1578 		while (srecmap[idx].uid) {
1579 			if (srecmap[idx].uid == uid) {
1580 				srec = srecmap[idx].srec;
1581 				goto found;
1582 			}
1583 			if (++idx == hashsz)
1584 				idx = 0;
1585 		}
1586 		continue;
1587 	  found:
1588 		tmsg->srec = srec;
1589 		srec->msg[t] = tmsg;
1590 	}
1591 	free( srecmap );
1592 
1593 	if (!(svars->state[1-t] & ST_LOADED))
1594 		return;
1595 
1596 	for (t = 0; t < 2; t++) {
1597 		if (svars->uidval[t] != UIDVAL_BAD && svars->uidval[t] != svars->newuidval[t]) {
1598 			// This code checks whether the messages with known UIDs are actually the
1599 			// same messages, as recognized by their Message-IDs.
1600 			unsigned need = 0, got = 0;
1601 			debug( "trying to re-approve uid validity of %s\n", str_fn[t] );
1602 			for (srec = svars->srecs; srec; srec = srec->next) {
1603 				if (srec->status & S_DEAD)
1604 					continue;
1605 				need++;
1606 				if (!srec->msg[t])
1607 					continue;  // Message disappeared.
1608 				// Present paired messages require re-validation.
1609 				if (!srec->msg[t]->msgid)
1610 					continue;  // Messages without ID are useless for re-validation.
1611 				if (!srec->msg[1-t])
1612 					continue;  // Partner disappeared.
1613 				if (!srec->msg[1-t]->msgid || strcmp( srec->msg[F]->msgid, srec->msg[N]->msgid )) {
1614 					error( "Error: channel %s, %s box %s: UIDVALIDITY genuinely changed (at UID %u).\n",
1615 					       svars->chan->name, str_fn[t], svars->orig_name[t], srec->uid[t] );
1616 				  uvchg:
1617 					svars->ret |= SYNC_FAIL;
1618 					cancel_sync( svars );
1619 					return;
1620 				}
1621 				got++;
1622 			}
1623 			// We encountered no messages that contradict the hypothesis that the
1624 			// UIDVALIDITY change was spurious.
1625 			// If we got enough messages confirming the hypothesis, we just accept it.
1626 			// If there aren't quite enough messages, we check that at least 80% of
1627 			// those previously present are still there and confirm the hypothesis;
1628 			// this also covers the case of a box that was already empty.
1629 			if (got < 20 && got * 5 < need * 4) {
1630 				// Too few confirmed messages. This is very likely in the drafts folder.
1631 				// A proper fallback would be fetching more headers (which potentially need
1632 				// normalization) or the message body (which should be truncated for sanity)
1633 				// and comparing.
1634 				error( "Error: channel %s, %s box %s: Unable to recover from UIDVALIDITY change.\n",
1635 				       svars->chan->name, str_fn[t], svars->orig_name[t] );
1636 				goto uvchg;
1637 			}
1638 			notice( "Notice: channel %s, %s box %s: Recovered from change of UIDVALIDITY.\n",
1639 			        svars->chan->name, str_fn[t], svars->orig_name[t] );
1640 			svars->uidval[t] = UIDVAL_BAD;
1641 		}
1642 	}
1643 
1644 	if (svars->uidval[F] == UIDVAL_BAD || svars->uidval[N] == UIDVAL_BAD) {
1645 		svars->uidval[F] = svars->newuidval[F];
1646 		svars->uidval[N] = svars->newuidval[N];
1647 		JLOG( "| %u %u", (svars->uidval[F], svars->uidval[N]), "new UIDVALIDITYs" );
1648 	}
1649 
1650 	svars->oldmaxuid[F] = svars->maxuid[F];
1651 	svars->oldmaxuid[N] = svars->maxuid[N];
1652 	svars->oldmaxxfuid = svars->maxxfuid;
1653 
1654 	info( "Synchronizing...\n" );
1655 	for (t = 0; t < 2; t++)
1656 		svars->good_flags[t] = (uchar)svars->drv[t]->get_supported_flags( svars->ctx[t] );
1657 
1658 	int any_new[2] = { 0, 0 };
1659 
1660 	debug( "synchronizing old entries\n" );
1661 	for (srec = svars->srecs; srec; srec = srec->next) {
1662 		if (srec->status & S_DEAD)
1663 			continue;
1664 		debug( "pair (%u,%u)\n", srec->uid[F], srec->uid[N] );
1665 		assert( !srec->tuid[0] );
1666 		// no[] means that a message is known to be not there.
1667 		no[F] = !srec->msg[F] && (svars->opts[F] & OPEN_OLD);
1668 		no[N] = !srec->msg[N] && (svars->opts[N] & OPEN_OLD);
1669 		if (no[F] && no[N]) {
1670 			// It does not matter whether one side was already known to be missing
1671 			// (never stored [skipped or failed] or expunged [possibly expired]) -
1672 			// now both are missing, so the entry is superfluous.
1673 			srec->status = S_DEAD;
1674 			JLOG( "- %u %u", (srec->uid[F], srec->uid[N]), "both missing" );
1675 		} else {
1676 			// del[] means that a message becomes known to have been expunged.
1677 			del[F] = no[F] && srec->uid[F];
1678 			del[N] = no[N] && srec->uid[N];
1679 
1680 			for (t = 0; t < 2; t++) {
1681 				if (srec->msg[t] && (srec->msg[t]->flags & F_DELETED))
1682 					srec->wstate |= W_DEL(t);
1683 				if (del[t]) {
1684 					// The target was newly expunged, so there is nothing to update.
1685 					// The deletion is propagated in the opposite iteration.
1686 				} else if (!srec->uid[t]) {
1687 					// The target was never stored, or was previously expunged, so there
1688 					// is nothing to update.
1689 					// Note: the opposite UID must be valid, as otherwise the entry would
1690 					// have been pruned already.
1691 				} else if (del[1-t]) {
1692 					// The source was newly expunged, so possibly propagate the deletion.
1693 					// The target may be in an unknown state (not fetched).
1694 					if ((t == F) && (srec->status & (S_EXPIRE|S_EXPIRED))) {
1695 						/* Don't propagate deletion resulting from expiration. */
1696 						JLOG( "> %u %u 0", (srec->uid[F], srec->uid[N]), "near side expired, orphaning far side" );
1697 						srec->uid[N] = 0;
1698 					} else {
1699 						if (srec->msg[t] && (srec->msg[t]->status & M_FLAGS) &&
1700 						    // Ignore deleted flag, as that's what we'll change ourselves ...
1701 						    (((srec->msg[t]->flags & ~F_DELETED) != (srec->flags & ~F_DELETED)) ||
1702 						     // ... except for undeletion, as that's the opposite.
1703 						     (!(srec->msg[t]->flags & F_DELETED) && (srec->flags & F_DELETED))))
1704 							notice( "Notice: conflicting changes in (%u,%u)\n", srec->uid[F], srec->uid[N] );
1705 						if (svars->chan->ops[t] & OP_DELETE) {
1706 							debug( "  %sing delete\n", str_hl[t] );
1707 							srec->aflags[t] = F_DELETED;
1708 							srec->wstate |= W_DELETE;
1709 						} else {
1710 							debug( "  not %sing delete\n", str_hl[t] );
1711 						}
1712 					}
1713 				} else if (!srec->msg[1-t]) {
1714 					// We have no source to work with, because it was never stored,
1715 					// it was previously expunged, or we did not fetch it.
1716 					debug( "  no %s\n", str_fn[1-t] );
1717 				} else {
1718 					// We have a source. The target may be in an unknown state.
1719 					if (svars->chan->ops[t] & OP_FLAGS) {
1720 						sflags = sanitize_flags( srec->msg[1-t]->flags, svars, t );
1721 						if ((t == F) && (srec->status & (S_EXPIRE|S_EXPIRED))) {
1722 							/* Don't propagate deletion resulting from expiration. */
1723 							debug( "  near side expiring\n" );
1724 							sflags &= ~F_DELETED;
1725 						}
1726 						if (srec->status & S_DUMMY(1-t)) {
1727 							// For placeholders, don't propagate:
1728 							// - Seen, because the real contents were obviously not seen yet
1729 							// - Flagged, because it's just a request to upgrade
1730 							sflags &= ~(F_SEEN|F_FLAGGED);
1731 						}
1732 						srec->aflags[t] = sflags & ~srec->flags;
1733 						srec->dflags[t] = ~sflags & srec->flags;
1734 						if ((DFlags & DEBUG_SYNC) && (srec->aflags[t] || srec->dflags[t])) {
1735 							char afbuf[16], dfbuf[16]; /* enlarge when support for keywords is added */
1736 							make_flags( srec->aflags[t], afbuf );
1737 							make_flags( srec->dflags[t], dfbuf );
1738 							debug( "  %sing flags: +%s -%s\n", str_hl[t], afbuf, dfbuf );
1739 						}
1740 					}
1741 				}
1742 			}
1743 
1744 			sync_rec_t *nsrec = srec;
1745 			if (((srec->status & S_DUMMY(F)) && (svars->chan->ops[F] & OP_RENEW)) ||
1746 			     ((srec->status & S_DUMMY(N)) && (svars->chan->ops[N] & OP_RENEW))) {
1747 				// Flagging the message on either side causes an upgrade of the dummy.
1748 				// We ignore flag resets, because that corner case is not worth it.
1749 				ushort muflags = srec->msg[F] ? srec->msg[F]->flags : 0;
1750 				ushort suflags = srec->msg[N] ? srec->msg[N]->flags : 0;
1751 				if ((muflags | suflags) & F_FLAGGED) {
1752 					t = (srec->status & S_DUMMY(F)) ? F : N;
1753 					// We calculate the flags for the replicated message already now,
1754 					// because after an interruption the dummy may be already gone.
1755 					srec->pflags = ((srec->msg[t]->flags & ~(F_SEEN|F_FLAGGED)) | srec->aflags[t]) & ~srec->dflags[t];
1756 					// Consequently, the srec's flags are committed right away as well.
1757 					srec->flags = (srec->flags | srec->aflags[t]) & ~srec->dflags[t];
1758 					JLOG( "^ %u %u %u %u", (srec->uid[F], srec->uid[N], srec->pflags, srec->flags), "upgrading placeholder" );
1759 					nsrec = upgrade_srec( svars, srec );
1760 				}
1761 			}
1762 			// This is separated, because the upgrade can come from the journal.
1763 			if (srec->wstate & W_UPGRADE) {
1764 				t = !srec->uid[F] ? F : N;
1765 				tmsg = srec->msg[1-t];
1766 				if ((svars->chan->ops[t] & OP_EXPUNGE) && (srec->pflags & F_DELETED)) {
1767 					JLOG( "- %u %u", (srec->uid[F], srec->uid[N]), "killing upgrade - would be expunged anyway" );
1768 					tmsg->srec = NULL;
1769 					srec->status = S_DEAD;
1770 				} else {
1771 					// Pretend that the source message has the adjusted flags of the dummy.
1772 					tmsg->flags = srec->pflags;
1773 					tmsg->status = M_FLAGS;
1774 					any_new[t] = 1;
1775 				}
1776 			}
1777 			srec = nsrec;
1778 		}
1779 	}
1780 
1781 	for (t = 0; t < 2; t++) {
1782 		debug( "synchronizing new messages on %s\n", str_fn[1-t] );
1783 		for (tmsg = svars->msgs[1-t]; tmsg; tmsg = tmsg->next) {
1784 			srec = tmsg->srec;
1785 			if (srec) {
1786 				if (srec->status & S_SKIPPED) {
1787 					// Pre-1.4 legacy only: The message was skipped due to being too big.
1788 					// We must have already seen the UID, but we might have been interrupted.
1789 					if (svars->maxuid[1-t] < tmsg->uid)
1790 						svars->maxuid[1-t] = tmsg->uid;
1791 					if (!(svars->chan->ops[t] & OP_RENEW))
1792 						continue;
1793 					srec->status = S_PENDING;
1794 					// The message size was not queried, so this won't be dummified below.
1795 					if (!(tmsg->flags & F_FLAGGED)) {
1796 						srec->status |= S_DUMMY(t);
1797 						JLOG( "_ %u %u", (srec->uid[F], srec->uid[N]), "placeholder only - was previously skipped" );
1798 					} else {
1799 						JLOG( "~ %u %u %u", (srec->uid[F], srec->uid[N], srec->status), "was previously skipped" );
1800 					}
1801 				} else {
1802 					if (!(svars->chan->ops[t] & OP_NEW))
1803 						continue;
1804 					// This catches messages:
1805 					// - that are actually new
1806 					// - whose propagation got interrupted
1807 					// - whose propagation was completed, but not logged yet
1808 					// - that aren't actually new, but a result of syncing, and the instant
1809 					//   maxuid upping was prevented by the presence of actually new messages
1810 					if (svars->maxuid[1-t] < tmsg->uid)
1811 						svars->maxuid[1-t] = tmsg->uid;
1812 					if (!(srec->status & S_PENDING))
1813 						continue;  // Nothing to do - the message is paired or expired
1814 					// Propagation was scheduled, but we got interrupted
1815 					debug( "unpropagated old message %u\n", tmsg->uid );
1816 				}
1817 
1818 				if ((svars->chan->ops[t] & OP_EXPUNGE) && (tmsg->flags & F_DELETED)) {
1819 					JLOG( "- %u %u", (srec->uid[F], srec->uid[N]), "killing - would be expunged anyway" );
1820 					tmsg->srec = NULL;
1821 					srec->status = S_DEAD;
1822 					continue;
1823 				}
1824 			} else {
1825 				if (!(svars->chan->ops[t] & OP_NEW))
1826 					continue;
1827 				if (tmsg->uid <= svars->maxuid[1-t]) {
1828 					// The message should be already paired. It's not, so it was:
1829 					// - previously paired, but the entry was expired and pruned => ignore
1830 					// - attempted, but failed => ignore (the wisdom of this is debatable)
1831 					// - ignored, as it would have been expunged anyway => ignore (even if undeleted)
1832 					continue;
1833 				}
1834 				svars->maxuid[1-t] = tmsg->uid;
1835 				debug( "new message %u\n", tmsg->uid );
1836 
1837 				if ((svars->chan->ops[t] & OP_EXPUNGE) && (tmsg->flags & F_DELETED)) {
1838 					debug( "-> ignoring - would be expunged anyway\n" );
1839 					continue;
1840 				}
1841 
1842 				srec = nfcalloc( sizeof(*srec) );
1843 				*svars->srecadd = srec;
1844 				svars->srecadd = &srec->next;
1845 				svars->nsrecs++;
1846 				srec->status = S_PENDING;
1847 				srec->uid[1-t] = tmsg->uid;
1848 				srec->msg[1-t] = tmsg;
1849 				tmsg->srec = srec;
1850 				JLOG( "+ %u %u", (srec->uid[F], srec->uid[N]), "fresh" );
1851 			}
1852 			if (!(tmsg->flags & F_FLAGGED) && tmsg->size > svars->chan->stores[t]->max_size &&
1853 			    !(srec->wstate & W_UPGRADE) && !(srec->status & (S_DUMMY(F)|S_DUMMY(N)))) {
1854 				srec->status |= S_DUMMY(t);
1855 				JLOG( "_ %u %u", (srec->uid[F], srec->uid[N]), "placeholder only - too big" );
1856 			}
1857 			any_new[t] = 1;
1858 		}
1859 	}
1860 
1861 	if ((svars->chan->ops[N] & (OP_NEW|OP_RENEW|OP_FLAGS)) && svars->chan->max_messages) {
1862 		// Note: When this branch is entered, we have loaded all near side messages.
1863 		/* Expire excess messages. Important (flagged, unread, or unpropagated) messages
1864 		 * older than the first not expired message are not counted towards the total. */
1865 		debug( "preparing message expiration\n" );
1866 		// Due to looping only over the far side, we completely ignore unpaired
1867 		// near-side messages. This is correct, as we cannot expire them without
1868 		// data loss anyway; consequently, we also don't count them.
1869 		// Note that we also ignore near-side messages we're currently propagating,
1870 		// which delays expiration of some messages by one cycle. Otherwise, we'd have
1871 		// to sequence flag propagation after message propagation to avoid a race
1872 		// with 3rd-party expunging, and that seems unreasonably expensive.
1873 		alive = 0;
1874 		for (tmsg = svars->msgs[F]; tmsg; tmsg = tmsg->next) {
1875 			if (tmsg->status & M_DEAD)
1876 				continue;
1877 			// We ignore unpaired far-side messages, as there is obviously nothing
1878 			// to expire in the first place.
1879 			if (!(srec = tmsg->srec))
1880 				continue;
1881 			if (!(srec->status & S_PENDING)) {
1882 				if (!srec->msg[N])
1883 					continue;  // Already expired or skipped.
1884 				nflags = (srec->msg[N]->flags | srec->aflags[N]) & ~srec->dflags[N];
1885 			} else {
1886 				nflags = tmsg->flags;
1887 			}
1888 			if (!(nflags & F_DELETED) || (srec->status & (S_EXPIRE|S_EXPIRED)))
1889 				// The message is not deleted, or it is, but only due to being expired.
1890 				alive++;
1891 		}
1892 		todel = alive - svars->chan->max_messages;
1893 		debug( "%d alive messages, %d excess - expiring\n", alive, todel );
1894 		alive = 0;
1895 		for (tmsg = svars->msgs[F]; tmsg; tmsg = tmsg->next) {
1896 			if (tmsg->status & M_DEAD)
1897 				continue;
1898 			if (!(srec = tmsg->srec))
1899 				continue;
1900 			if (!(srec->status & S_PENDING)) {
1901 				if (!srec->msg[N])
1902 					continue;
1903 				nflags = (srec->msg[N]->flags | srec->aflags[N]) & ~srec->dflags[N];
1904 			} else {
1905 				nflags = tmsg->flags;
1906 			}
1907 			if (!(nflags & F_DELETED) || (srec->status & (S_EXPIRE|S_EXPIRED))) {
1908 				if ((nflags & F_FLAGGED) ||
1909 				    !((nflags & F_SEEN) || ((void)(todel > 0 && alive++), svars->chan->expire_unread > 0))) {
1910 					// Important messages are always fetched/kept.
1911 					debug( "  pair(%u,%u) is important\n", srec->uid[F], srec->uid[N] );
1912 					todel--;
1913 				} else if (todel > 0 ||
1914 				           ((srec->status & (S_EXPIRE|S_EXPIRED)) == (S_EXPIRE|S_EXPIRED)) ||
1915 				           ((srec->status & (S_EXPIRE|S_EXPIRED)) && (srec->msg[N]->flags & F_DELETED))) {
1916 					/* The message is excess or was already (being) expired. */
1917 					srec->wstate |= W_NEXPIRE;
1918 					debug( "  pair(%u,%u) expired\n", srec->uid[F], srec->uid[N] );
1919 					if (svars->maxxfuid < srec->uid[F])
1920 						svars->maxxfuid = srec->uid[F];
1921 					todel--;
1922 				}
1923 			}
1924 		}
1925 		debug( "%d excess messages remain\n", todel );
1926 		if (svars->chan->expire_unread < 0 && alive * 2 > svars->chan->max_messages) {
1927 			error( "%s: %d unread messages in excess of MaxMessages (%d).\n"
1928 			       "Please set ExpireUnread to decide outcome. Skipping mailbox.\n",
1929 			       svars->orig_name[N], alive, svars->chan->max_messages );
1930 			svars->ret |= SYNC_FAIL;
1931 			cancel_sync( svars );
1932 			return;
1933 		}
1934 		for (srec = svars->srecs; srec; srec = srec->next) {
1935 			if (srec->status & S_DEAD)
1936 				continue;
1937 			if (!(srec->status & S_PENDING)) {
1938 				if (!srec->msg[N])
1939 					continue;
1940 				uchar nex = (srec->wstate / W_NEXPIRE) & 1;
1941 				if (nex != ((srec->status / S_EXPIRED) & 1)) {
1942 					/* The record needs a state change ... */
1943 					if (nex != ((srec->status / S_EXPIRE) & 1)) {
1944 						/* ... and we need to start a transaction. */
1945 						srec->status = (srec->status & ~S_EXPIRE) | (nex * S_EXPIRE);
1946 						JLOG( "~ %u %u %u", (srec->uid[F], srec->uid[N], srec->status), "expire %u - begin", nex );
1947 					} else {
1948 						/* ... but the "right" transaction is already pending. */
1949 						debug( "-> pair(%u,%u): expire %u (pending)\n", srec->uid[F], srec->uid[N], nex );
1950 					}
1951 				} else {
1952 					/* Note: the "wrong" transaction may be pending here,
1953 					 * e.g.: W_NEXPIRE = 0, S_EXPIRE = 1, S_EXPIRED = 0. */
1954 				}
1955 			} else {
1956 				if (srec->wstate & W_NEXPIRE) {
1957 					JLOG( "= %u %u", (srec->uid[F], srec->uid[N]), "expire unborn" );
1958 					// If we have so many new messages that some of them are instantly expired,
1959 					// but some are still propagated because they are important, we need to
1960 					// ensure explicitly that the bulk fetch limit is upped.
1961 					if (svars->maxxfuid < srec->uid[F])
1962 						svars->maxxfuid = srec->uid[F];
1963 					srec->msg[F]->srec = NULL;
1964 					srec->status = S_DEAD;
1965 				}
1966 			}
1967 		}
1968 	}
1969 
1970 	sync_ref( svars );
1971 
1972 	debug( "synchronizing flags\n" );
1973 	for (srec = svars->srecs; srec; srec = srec->next) {
1974 		if (srec->status & S_DEAD)
1975 			continue;
1976 		for (t = 0; t < 2; t++) {
1977 			if (!srec->uid[t])
1978 				continue;
1979 			aflags = srec->aflags[t];
1980 			dflags = srec->dflags[t];
1981 			if (srec->wstate & (W_DELETE|W_PURGE)) {
1982 				if (!aflags) {
1983 					// This deletion propagation goes the other way round, or
1984 					// this deletion of a dummy happens on the other side.
1985 					continue;
1986 				}
1987 				if (!srec->msg[t] && (svars->opts[t] & OPEN_OLD)) {
1988 					// The message disappeared. This can happen, because the wstate may
1989 					// come from the journal, and things could have happened meanwhile.
1990 					continue;
1991 				}
1992 			} else {
1993 				/* The trigger is an expiration transaction being ongoing ... */
1994 				if ((t == N) && ((shifted_bit(srec->status, S_EXPIRE, S_EXPIRED) ^ srec->status) & S_EXPIRED)) {
1995 					/* ... but the actual action derives from the wanted state. */
1996 					if (srec->wstate & W_NEXPIRE)
1997 						aflags |= F_DELETED;
1998 					else
1999 						dflags |= F_DELETED;
2000 				}
2001 			}
2002 			if ((svars->chan->ops[t] & OP_EXPUNGE) && (((srec->msg[t] ? srec->msg[t]->flags : 0) | aflags) & ~dflags & F_DELETED) &&
2003 			    (!svars->ctx[t]->conf->trash || svars->ctx[t]->conf->trash_only_new))
2004 			{
2005 				/* If the message is going to be expunged, don't propagate anything but the deletion. */
2006 				srec->aflags[t] &= F_DELETED;
2007 				aflags &= F_DELETED;
2008 				srec->dflags[t] = dflags = 0;
2009 			}
2010 			if (srec->msg[t] && (srec->msg[t]->status & M_FLAGS)) {
2011 				/* If we know the target message's state, optimize away non-changes. */
2012 				aflags &= ~srec->msg[t]->flags;
2013 				dflags &= srec->msg[t]->flags;
2014 			}
2015 			if (aflags | dflags) {
2016 				flags_total[t]++;
2017 				stats();
2018 				svars->flags_pending[t]++;
2019 				fv = nfmalloc( sizeof(*fv) );
2020 				fv->aux = AUX;
2021 				fv->srec = srec;
2022 				fv->aflags = aflags;
2023 				fv->dflags = dflags;
2024 				svars->drv[t]->set_msg_flags( svars->ctx[t], srec->msg[t], srec->uid[t], aflags, dflags, flags_set, fv );
2025 				if (check_cancel( svars ))
2026 					goto out;
2027 			} else
2028 				flags_set_p2( svars, srec, t );
2029 		}
2030 	}
2031 	for (t = 0; t < 2; t++) {
2032 		svars->drv[t]->commit_cmds( svars->ctx[t] );
2033 		svars->state[t] |= ST_SENT_FLAGS;
2034 		msgs_flags_set( svars, t );
2035 		if (check_cancel( svars ))
2036 			goto out;
2037 	}
2038 
2039 	debug( "propagating new messages\n" );
2040 	if (UseFSync && svars->jfp)
2041 		fdatasync( fileno( svars->jfp ) );
2042 	for (t = 0; t < 2; t++) {
2043 		if (any_new[t]) {
2044 			svars->finduid[t] = svars->drv[t]->get_uidnext( svars->ctx[t] );
2045 			JLOG( "F %d %u", (t, svars->finduid[t]), "save UIDNEXT of %s", str_fn[t] );
2046 			svars->new_msgs[t] = svars->msgs[1-t];
2047 		} else {
2048 			svars->state[t] |= ST_SENT_NEW;
2049 		}
2050 		msgs_copied( svars, t );
2051 		if (check_cancel( svars ))
2052 			goto out;
2053 	}
2054 
2055   out:
2056 	sync_deref( svars );
2057 }
2058 
2059 static void
msg_copied(int sts,uint uid,copy_vars_t * vars)2060 msg_copied( int sts, uint uid, copy_vars_t *vars )
2061 {
2062 	SVARS_CHECK_CANCEL_RET;
2063 	sync_rec_t *srec = vars->srec;
2064 	switch (sts) {
2065 	case SYNC_OK:
2066 		if (!(srec->wstate & W_UPGRADE) && vars->msg->flags != srec->flags) {
2067 			srec->flags = vars->msg->flags;
2068 			JLOG( "* %u %u %u", (srec->uid[F], srec->uid[N], srec->flags), "%sed with flags", str_hl[t] );
2069 		}
2070 		if (!uid) {  // Stored to a non-UIDPLUS mailbox
2071 			svars->state[t] |= ST_FIND_NEW;
2072 		} else {
2073 			ASSIGN_UID( srec, t, uid, "%sed message", str_hl[t] );
2074 		}
2075 		break;
2076 	case SYNC_NOGOOD:
2077 		srec->status = S_DEAD;
2078 		JLOG( "- %u %u", (srec->uid[F], srec->uid[N]), "%s failed", str_hl[t] );
2079 		break;
2080 	default:
2081 		cancel_sync( svars );
2082 		free( vars );
2083 		return;
2084 	}
2085 	free( vars );
2086 	new_done[t]++;
2087 	stats();
2088 	svars->new_pending[t]--;
2089 	msgs_copied( svars, t );
2090 }
2091 
2092 static void msgs_found_new( int sts, message_t *msgs, void *aux );
2093 static void msgs_new_done( sync_vars_t *svars, int t );
2094 static void sync_close( sync_vars_t *svars, int t );
2095 
2096 static void
msgs_copied(sync_vars_t * svars,int t)2097 msgs_copied( sync_vars_t *svars, int t )
2098 {
2099 	message_t *tmsg;
2100 	sync_rec_t *srec;
2101 	copy_vars_t *cv;
2102 
2103 	if (svars->state[t] & ST_SENDING_NEW)
2104 		return;
2105 
2106 	sync_ref( svars );
2107 
2108 	if (!(svars->state[t] & ST_SENT_NEW)) {
2109 		for (tmsg = svars->new_msgs[t]; tmsg; tmsg = tmsg->next) {
2110 			if ((srec = tmsg->srec) && (srec->status & S_PENDING)) {
2111 				if (svars->drv[t]->get_memory_usage( svars->ctx[t] ) >= BufferLimit) {
2112 					svars->new_msgs[t] = tmsg;
2113 					goto out;
2114 				}
2115 				for (uint i = 0; i < TUIDL; i++) {
2116 					uchar c = arc4_getbyte() & 0x3f;
2117 					srec->tuid[i] = (char)(c < 26 ? c + 'A' : c < 52 ? c + 'a' - 26 : c < 62 ? c + '0' - 52 : c == 62 ? '+' : '/');
2118 				}
2119 				JLOG( "# %u %u %." stringify(TUIDL) "s", (srec->uid[F], srec->uid[N], srec->tuid), "%sing message", str_hl[t] );
2120 				new_total[t]++;
2121 				stats();
2122 				svars->new_pending[t]++;
2123 				svars->state[t] |= ST_SENDING_NEW;
2124 				cv = nfmalloc( sizeof(*cv) );
2125 				cv->cb = msg_copied;
2126 				cv->aux = AUX;
2127 				cv->srec = srec;
2128 				cv->msg = tmsg;
2129 				cv->minimal = (srec->status & S_DUMMY(t));
2130 				copy_msg( cv );
2131 				svars->state[t] &= ~ST_SENDING_NEW;
2132 				if (check_cancel( svars ))
2133 					goto out;
2134 			}
2135 		}
2136 		svars->state[t] |= ST_SENT_NEW;
2137 	}
2138 
2139 	if (svars->new_pending[t])
2140 		goto out;
2141 
2142 	sync_close( svars, 1-t );
2143 	if (check_cancel( svars ))
2144 		goto out;
2145 
2146 	if (svars->state[t] & ST_FIND_NEW) {
2147 		debug( "finding just copied messages on %s\n", str_fn[t] );
2148 		svars->drv[t]->find_new_msgs( svars->ctx[t], svars->finduid[t], msgs_found_new, AUX );
2149 	} else {
2150 		msgs_new_done( svars, t );
2151 	}
2152 
2153   out:
2154 	sync_deref( svars );
2155 }
2156 
2157 static void
msgs_found_new(int sts,message_t * msgs,void * aux)2158 msgs_found_new( int sts, message_t *msgs, void *aux )
2159 {
2160 	SVARS_CHECK_RET;
2161 	switch (sts) {
2162 	case DRV_OK:
2163 		debug( "matching just copied messages on %s\n", str_fn[t] );
2164 		break;
2165 	default:
2166 		warn( "Warning: cannot find newly stored messages on %s.\n", str_fn[t] );
2167 		break;
2168 	}
2169 	match_tuids( svars, t, msgs );
2170 	msgs_new_done( svars, t );
2171 }
2172 
2173 static void
msgs_new_done(sync_vars_t * svars,int t)2174 msgs_new_done( sync_vars_t *svars, int t )
2175 {
2176 	svars->state[t] |= ST_FOUND_NEW;
2177 	sync_close( svars, t );
2178 }
2179 
2180 static void
flags_set(int sts,void * aux)2181 flags_set( int sts, void *aux )
2182 {
2183 	SVARS_CHECK_RET_VARS(flag_vars_t);
2184 	sync_rec_t *srec = vars->srec;
2185 	switch (sts) {
2186 	case DRV_OK:
2187 		if (vars->aflags & F_DELETED)
2188 			srec->wstate |= W_DEL(t);
2189 		else if (vars->dflags & F_DELETED)
2190 			srec->wstate &= ~W_DEL(t);
2191 		flags_set_p2( svars, srec, t );
2192 		break;
2193 	}
2194 	free( vars );
2195 	flags_done[t]++;
2196 	stats();
2197 	svars->flags_pending[t]--;
2198 	msgs_flags_set( svars, t );
2199 }
2200 
2201 static void
flags_set_p2(sync_vars_t * svars,sync_rec_t * srec,int t)2202 flags_set_p2( sync_vars_t *svars, sync_rec_t *srec, int t )
2203 {
2204 	if (srec->wstate & W_DELETE) {
2205 		JLOG( "%c %u %u 0", ("><"[t], srec->uid[F], srec->uid[N]), "%sed deletion", str_hl[t] );
2206 		srec->uid[1-t] = 0;
2207 	} else {
2208 		uchar nflags = (srec->flags | srec->aflags[t]) & ~srec->dflags[t];
2209 		if (srec->flags != nflags) {
2210 			JLOG( "* %u %u %u", (srec->uid[F], srec->uid[N], nflags), "%sed flags; were %u", (str_hl[t], srec->flags) );
2211 			srec->flags = nflags;
2212 		}
2213 		if (t == N) {
2214 			uchar nex = (srec->wstate / W_NEXPIRE) & 1;
2215 			if (nex != ((srec->status / S_EXPIRED) & 1)) {
2216 				srec->status = (srec->status & ~S_EXPIRED) | (nex * S_EXPIRED);
2217 				JLOG( "~ %u %u %u", (srec->uid[F], srec->uid[N], srec->status), "expired %d - commit", nex );
2218 			} else if (nex != ((srec->status / S_EXPIRE) & 1)) {
2219 				srec->status = (srec->status & ~S_EXPIRE) | (nex * S_EXPIRE);
2220 				JLOG( "~ %u %u %u", (srec->uid[F], srec->uid[N], srec->status), "expire %d - cancel", nex );
2221 			}
2222 		}
2223 	}
2224 }
2225 
2226 typedef struct {
2227 	void *aux;
2228 	message_t *msg;
2229 } trash_vars_t;
2230 
2231 static void msg_trashed( int sts, void *aux );
2232 static void msg_rtrashed( int sts, uint uid, copy_vars_t *vars );
2233 
2234 static void
msgs_flags_set(sync_vars_t * svars,int t)2235 msgs_flags_set( sync_vars_t *svars, int t )
2236 {
2237 	message_t *tmsg;
2238 	trash_vars_t *tv;
2239 	copy_vars_t *cv;
2240 
2241 	if (!(svars->state[t] & ST_SENT_FLAGS) || svars->flags_pending[t])
2242 		return;
2243 
2244 	sync_ref( svars );
2245 
2246 	if ((svars->chan->ops[t] & OP_EXPUNGE) &&
2247 	    (svars->ctx[t]->conf->trash || (svars->ctx[1-t]->conf->trash && svars->ctx[1-t]->conf->trash_remote_new))) {
2248 		debug( "trashing on %s\n", str_fn[t] );
2249 		for (tmsg = svars->msgs[t]; tmsg; tmsg = tmsg->next)
2250 			if ((tmsg->flags & F_DELETED) && !find_uint_array( svars->trashed_msgs[t].array, tmsg->uid ) &&
2251 			    (t == F || !tmsg->srec || !(tmsg->srec->status & (S_EXPIRE|S_EXPIRED)))) {
2252 				if (svars->ctx[t]->conf->trash) {
2253 					if (!svars->ctx[t]->conf->trash_only_new || !tmsg->srec || (tmsg->srec->status & (S_PENDING | S_SKIPPED))) {
2254 						debug( "%s: trashing message %u\n", str_fn[t], tmsg->uid );
2255 						trash_total[t]++;
2256 						stats();
2257 						svars->trash_pending[t]++;
2258 						tv = nfmalloc( sizeof(*tv) );
2259 						tv->aux = AUX;
2260 						tv->msg = tmsg;
2261 						svars->drv[t]->trash_msg( svars->ctx[t], tmsg, msg_trashed, tv );
2262 						if (check_cancel( svars ))
2263 							goto out;
2264 					} else
2265 						debug( "%s: not trashing message %u - not new\n", str_fn[t], tmsg->uid );
2266 				} else {
2267 					if (!tmsg->srec || (tmsg->srec->status & (S_PENDING | S_SKIPPED))) {
2268 						if (tmsg->size <= svars->ctx[1-t]->conf->max_size) {
2269 							debug( "%s: remote trashing message %u\n", str_fn[t], tmsg->uid );
2270 							trash_total[t]++;
2271 							stats();
2272 							svars->trash_pending[t]++;
2273 							cv = nfmalloc( sizeof(*cv) );
2274 							cv->cb = msg_rtrashed;
2275 							cv->aux = INV_AUX;
2276 							cv->srec = NULL;
2277 							cv->msg = tmsg;
2278 							cv->minimal = 0;
2279 							copy_msg( cv );
2280 							if (check_cancel( svars ))
2281 								goto out;
2282 						} else
2283 							debug( "%s: not remote trashing message %u - too big\n", str_fn[t], tmsg->uid );
2284 					} else
2285 						debug( "%s: not remote trashing message %u - not new\n", str_fn[t], tmsg->uid );
2286 				}
2287 			}
2288 	}
2289 	svars->state[t] |= ST_SENT_TRASH;
2290 	sync_close( svars, t );
2291 
2292   out:
2293 	sync_deref( svars );
2294 }
2295 
2296 static void
msg_trashed(int sts,void * aux)2297 msg_trashed( int sts, void *aux )
2298 {
2299 	trash_vars_t *vars = (trash_vars_t *)aux;
2300 	DECL_SVARS;
2301 
2302 	if (sts == DRV_MSG_BAD)
2303 		sts = DRV_BOX_BAD;
2304 	if (check_ret( sts, vars->aux ))
2305 		return;
2306 	INIT_SVARS(vars->aux);
2307 	JLOG( "T %d %u", (t, vars->msg->uid), "trashed on %s", str_fn[t] );
2308 	free( vars );
2309 	trash_done[t]++;
2310 	stats();
2311 	svars->trash_pending[t]--;
2312 	sync_close( svars, t );
2313 }
2314 
2315 static void
msg_rtrashed(int sts,uint uid ATTR_UNUSED,copy_vars_t * vars)2316 msg_rtrashed( int sts, uint uid ATTR_UNUSED, copy_vars_t *vars )
2317 {
2318 	SVARS_CHECK_CANCEL_RET;
2319 	switch (sts) {
2320 	case SYNC_OK:
2321 	case SYNC_NOGOOD: /* the message is gone or heavily busted */
2322 		break;
2323 	default:
2324 		cancel_sync( svars );
2325 		free( vars );
2326 		return;
2327 	}
2328 	t ^= 1;
2329 	JLOG( "T %d %u", (t, vars->msg->uid), "trashed remotely on %s", str_fn[1-t] );
2330 	free( vars );
2331 	trash_done[t]++;
2332 	stats();
2333 	svars->trash_pending[t]--;
2334 	sync_close( svars, t );
2335 }
2336 
2337 static void box_closed( int sts, void *aux );
2338 static void box_closed_p2( sync_vars_t *svars, int t );
2339 
2340 static void
sync_close(sync_vars_t * svars,int t)2341 sync_close( sync_vars_t *svars, int t )
2342 {
2343 	if ((~svars->state[t] & (ST_FOUND_NEW|ST_SENT_TRASH)) || svars->trash_pending[t] ||
2344 	    !(svars->state[1-t] & ST_SENT_NEW) || svars->new_pending[1-t])
2345 		return;
2346 
2347 	if (svars->state[t] & ST_CLOSING)
2348 		return;
2349 	svars->state[t] |= ST_CLOSING;
2350 
2351 	if ((svars->chan->ops[t] & OP_EXPUNGE) /*&& !(svars->state[t] & ST_TRASH_BAD)*/) {
2352 		debug( "expunging %s\n", str_fn[t] );
2353 		svars->drv[t]->close_box( svars->ctx[t], box_closed, AUX );
2354 	} else {
2355 		box_closed_p2( svars, t );
2356 	}
2357 }
2358 
2359 static void
box_closed(int sts,void * aux)2360 box_closed( int sts, void *aux )
2361 {
2362 	SVARS_CHECK_RET;
2363 	svars->state[t] |= ST_DID_EXPUNGE;
2364 	box_closed_p2( svars, t );
2365 }
2366 
2367 static void
box_closed_p2(sync_vars_t * svars,int t)2368 box_closed_p2( sync_vars_t *svars, int t )
2369 {
2370 	sync_rec_t *srec;
2371 
2372 	svars->state[t] |= ST_CLOSED;
2373 	if (!(svars->state[1-t] & ST_CLOSED))
2374 		return;
2375 
2376 	// All the journalling done in this function is merely for the autotest -
2377 	// the operations are idempotent, and we're about to commit the new state
2378 	// right afterwards anyway.
2379 
2380 	for (t = 0; t < 2; t++) {
2381 		// Committing maxuid is delayed until all messages were propagated, to
2382 		// ensure that all pending messages are still loaded next time in case
2383 		// of interruption - in particular skipping big messages would otherwise
2384 		// up the limit too early.
2385 		if (svars->maxuid[t] != svars->oldmaxuid[t])
2386 			JLOG( "N %d %u", (t, svars->maxuid[t]), "up maxuid of %s", str_fn[t] );
2387 	}
2388 
2389 	if (((svars->state[F] | svars->state[N]) & ST_DID_EXPUNGE) || svars->chan->max_messages) {
2390 		debug( "purging obsolete entries\n" );
2391 		for (srec = svars->srecs; srec; srec = srec->next) {
2392 			if (srec->status & S_DEAD)
2393 				continue;
2394 			if (!srec->uid[N] || ((srec->wstate & W_DEL(N)) && (svars->state[N] & ST_DID_EXPUNGE))) {
2395 				if (!srec->uid[F] || ((srec->wstate & W_DEL(F)) && (svars->state[F] & ST_DID_EXPUNGE)) ||
2396 				    ((srec->status & S_EXPIRED) && svars->maxuid[F] >= srec->uid[F] && svars->maxxfuid >= srec->uid[F])) {
2397 					JLOG( "- %u %u", (srec->uid[F], srec->uid[N]), "killing" );
2398 					srec->status = S_DEAD;
2399 				} else if (srec->uid[N]) {
2400 					JLOG( "> %u %u 0", (srec->uid[F], srec->uid[N]), "orphaning" );
2401 					srec->uid[N] = 0;
2402 				}
2403 			} else if (srec->uid[F] && ((srec->wstate & W_DEL(F)) && (svars->state[F] & ST_DID_EXPUNGE))) {
2404 				JLOG( "< %u %u 0", (srec->uid[F], srec->uid[N]), "orphaning" );
2405 				srec->uid[F] = 0;
2406 			}
2407 		}
2408 	}
2409 
2410 	// This is just an optimization, so it needs no journaling of intermediate states.
2411 	// However, doing it before the entry purge would require ensuring that the
2412 	// exception list includes all relevant messages.
2413 	if (svars->maxxfuid != svars->oldmaxxfuid)
2414 		JLOG( "! %u", svars->maxxfuid, "max expired UID on far side" );
2415 
2416 	save_state( svars );
2417 
2418 	sync_bail( svars );
2419 }
2420 
2421 static void
sync_bail(sync_vars_t * svars)2422 sync_bail( sync_vars_t *svars )
2423 {
2424 	sync_rec_t *srec, *nsrec;
2425 
2426 	free( svars->trashed_msgs[F].array.data );
2427 	free( svars->trashed_msgs[N].array.data );
2428 	for (srec = svars->srecs; srec; srec = nsrec) {
2429 		nsrec = srec->next;
2430 		free( srec );
2431 	}
2432 	if (svars->lfd >= 0) {
2433 		unlink( svars->lname );
2434 		close( svars->lfd );
2435 	}
2436 	sync_bail2( svars );
2437 }
2438 
2439 static void
sync_bail2(sync_vars_t * svars)2440 sync_bail2( sync_vars_t *svars )
2441 {
2442 	free( svars->lname );
2443 	free( svars->nname );
2444 	free( svars->jname );
2445 	free( svars->dname );
2446 	sync_bail3( svars );
2447 }
2448 
2449 static void
sync_bail3(sync_vars_t * svars)2450 sync_bail3( sync_vars_t *svars )
2451 {
2452 	free( svars->box_name[F] );
2453 	free( svars->box_name[N] );
2454 	sync_deref( svars );
2455 }
2456 
2457 static void
sync_deref(sync_vars_t * svars)2458 sync_deref( sync_vars_t *svars )
2459 {
2460 	if (!--svars->ref_count) {
2461 		void (*cb)( int sts, void *aux ) = svars->cb;
2462 		void *aux = svars->aux;
2463 		int ret = svars->ret;
2464 		free( svars );
2465 		cb( ret, aux );
2466 	}
2467 }
2468