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