sync.c 64 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258
  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. #include "sync.h"
  23. #include <assert.h>
  24. #include <stdio.h>
  25. #include <limits.h>
  26. #include <stdlib.h>
  27. #include <stddef.h>
  28. #include <unistd.h>
  29. #include <time.h>
  30. #include <fcntl.h>
  31. #include <ctype.h>
  32. #include <string.h>
  33. #include <errno.h>
  34. #include <sys/stat.h>
  35. #if !defined(_POSIX_SYNCHRONIZED_IO) || _POSIX_SYNCHRONIZED_IO <= 0
  36. # define fdatasync fsync
  37. #endif
  38. channel_conf_t global_conf;
  39. channel_conf_t *channels;
  40. group_conf_t *groups;
  41. const char *str_ms[] = { "master", "slave" }, *str_hl[] = { "push", "pull" };
  42. static void ATTR_PRINTFLIKE(1, 2)
  43. debug( const char *msg, ... )
  44. {
  45. va_list va;
  46. va_start( va, msg );
  47. vdebug( DEBUG_SYNC, msg, va );
  48. va_end( va );
  49. }
  50. static void ATTR_PRINTFLIKE(1, 2)
  51. debugn( const char *msg, ... )
  52. {
  53. va_list va;
  54. va_start( va, msg );
  55. vdebugn( DEBUG_SYNC, msg, va );
  56. va_end( va );
  57. }
  58. void
  59. Fclose( FILE *f, int safe )
  60. {
  61. if ((safe && (fflush( f ) || (UseFSync && fdatasync( fileno( f ) )))) || fclose( f ) == EOF) {
  62. sys_error( "Error: cannot close file" );
  63. exit( 1 );
  64. }
  65. }
  66. void
  67. vFprintf( FILE *f, const char *msg, va_list va )
  68. {
  69. int r;
  70. r = vfprintf( f, msg, va );
  71. if (r < 0) {
  72. sys_error( "Error: cannot write file" );
  73. exit( 1 );
  74. }
  75. }
  76. void
  77. Fprintf( FILE *f, const char *msg, ... )
  78. {
  79. va_list va;
  80. va_start( va, msg );
  81. vFprintf( f, msg, va );
  82. va_end( va );
  83. }
  84. static const char Flags[] = { 'D', 'F', 'R', 'S', 'T' };
  85. static int
  86. parse_flags( const char *buf )
  87. {
  88. uint flags, i, d;
  89. for (flags = i = d = 0; i < as(Flags); i++)
  90. if (buf[d] == Flags[i]) {
  91. flags |= (1 << i);
  92. d++;
  93. }
  94. return flags;
  95. }
  96. static int
  97. make_flags( int flags, char *buf )
  98. {
  99. uint i, d;
  100. for (i = d = 0; i < as(Flags); i++)
  101. if (flags & (1 << i))
  102. buf[d++] = Flags[i];
  103. buf[d] = 0;
  104. return d;
  105. }
  106. #define S_DEAD (1<<0) /* ephemeral: the entry was killed and should be ignored */
  107. #define S_DEL(ms) (1<<(2+(ms))) /* ephemeral: m/s message would be subject to expunge */
  108. #define S_EXPIRED (1<<4) /* the entry is expired (slave message removal confirmed) */
  109. #define S_EXPIRE (1<<5) /* the entry is being expired (slave message removal scheduled) */
  110. #define S_NEXPIRE (1<<6) /* temporary: new expiration state */
  111. #define S_DELETE (1<<7) /* ephemeral: flags propagation is a deletion */
  112. typedef struct sync_rec {
  113. struct sync_rec *next;
  114. /* string_list_t *keywords; */
  115. int uid[2]; /* -2 = pending (use tuid), -1 = skipped (too big), 0 = expired */
  116. message_t *msg[2];
  117. uchar status, flags, aflags[2], dflags[2];
  118. char tuid[TUIDL];
  119. } sync_rec_t;
  120. /* cases:
  121. a) both non-null
  122. b) only master null
  123. b.1) uid[M] 0
  124. b.2) uid[M] -1
  125. b.3) master not scanned
  126. b.4) master gone
  127. c) only slave null
  128. c.1) uid[S] 0
  129. c.2) uid[S] -1
  130. c.3) slave not scanned
  131. c.4) slave gone
  132. d) both null
  133. d.1) both gone
  134. d.2) uid[M] 0, slave not scanned
  135. d.3) uid[M] -1, slave not scanned
  136. d.4) master gone, slave not scanned
  137. d.5) uid[M] 0, slave gone
  138. d.6) uid[M] -1, slave gone
  139. d.7) uid[S] 0, master not scanned
  140. d.8) uid[S] -1, master not scanned
  141. d.9) slave gone, master not scanned
  142. d.10) uid[S] 0, master gone
  143. d.11) uid[S] -1, master gone
  144. impossible cases: both uid[M] & uid[S] 0 or -1, both not scanned
  145. */
  146. typedef struct {
  147. int t[2];
  148. void (*cb)( int sts, void *aux ), *aux;
  149. char *dname, *jname, *nname, *lname, *box_name[2];
  150. FILE *jfp, *nfp;
  151. sync_rec_t *srecs, **srecadd;
  152. channel_conf_t *chan;
  153. store_t *ctx[2];
  154. driver_t *drv[2];
  155. const char *orig_name[2];
  156. message_t *new_msgs[2];
  157. int_array_alloc_t trashed_msgs[2];
  158. int state[2], ref_count, nsrecs, ret, lfd, existing, replayed;
  159. int new_pending[2], flags_pending[2], trash_pending[2];
  160. int maxuid[2]; /* highest UID that was already propagated */
  161. int newmaxuid[2]; /* highest UID that is currently being propagated */
  162. int uidval[2]; /* UID validity value */
  163. int newuid[2]; /* TUID lookup makes sense only for UIDs >= this */
  164. int mmaxxuid; /* highest expired UID on master during new message propagation */
  165. int smaxxuid; /* highest expired UID on slave */
  166. } sync_vars_t;
  167. static void sync_ref( sync_vars_t *svars ) { ++svars->ref_count; }
  168. static void sync_deref( sync_vars_t *svars );
  169. static int check_cancel( sync_vars_t *svars );
  170. #define AUX &svars->t[t]
  171. #define INV_AUX &svars->t[1-t]
  172. #define DECL_SVARS \
  173. int t; \
  174. sync_vars_t *svars
  175. #define INIT_SVARS(aux) \
  176. t = *(int *)aux; \
  177. svars = (sync_vars_t *)(((char *)(&((int *)aux)[-t])) - offsetof(sync_vars_t, t))
  178. #define DECL_INIT_SVARS(aux) \
  179. int t = *(int *)aux; \
  180. sync_vars_t *svars = (sync_vars_t *)(((char *)(&((int *)aux)[-t])) - offsetof(sync_vars_t, t))
  181. /* operation dependencies:
  182. select(x): -
  183. load(x): select(x)
  184. new(M), new(S), flags(M), flags(S): load(M) & load(S)
  185. find_new(x): new(x)
  186. trash(x): flags(x)
  187. close(x): trash(x) & find_new(x) & new(!x) // with expunge
  188. cleanup: close(M) & close(S)
  189. */
  190. #define ST_LOADED (1<<0)
  191. #define ST_FIND_OLD (1<<1)
  192. #define ST_SENT_NEW (1<<2)
  193. #define ST_FIND_NEW (1<<3)
  194. #define ST_FOUND_NEW (1<<4)
  195. #define ST_SENT_FLAGS (1<<5)
  196. #define ST_SENT_TRASH (1<<6)
  197. #define ST_CLOSED (1<<7)
  198. #define ST_SENT_CANCEL (1<<8)
  199. #define ST_CANCELED (1<<9)
  200. #define ST_SELECTED (1<<10)
  201. #define ST_DID_EXPUNGE (1<<11)
  202. #define ST_CLOSING (1<<12)
  203. #define ST_CONFIRMED (1<<13)
  204. #define ST_PRESENT (1<<14)
  205. #define ST_SENDING_NEW (1<<15)
  206. void
  207. jFprintf( sync_vars_t *svars, const char *msg, ... )
  208. {
  209. va_list va;
  210. va_start( va, msg );
  211. vFprintf( svars->jfp, msg, va );
  212. va_end( va );
  213. }
  214. static void
  215. match_tuids( sync_vars_t *svars, int t )
  216. {
  217. sync_rec_t *srec;
  218. message_t *tmsg, *ntmsg = 0;
  219. const char *diag;
  220. int num_lost = 0;
  221. for (srec = svars->srecs; srec; srec = srec->next) {
  222. if (srec->status & S_DEAD)
  223. continue;
  224. if (srec->uid[t] == -2 && srec->tuid[0]) {
  225. debug( " pair(%d,%d): lookup %s, TUID %." stringify(TUIDL) "s\n", srec->uid[M], srec->uid[S], str_ms[t], srec->tuid );
  226. for (tmsg = ntmsg; tmsg; tmsg = tmsg->next) {
  227. if (tmsg->status & M_DEAD)
  228. continue;
  229. if (tmsg->tuid[0] && !memcmp( tmsg->tuid, srec->tuid, TUIDL )) {
  230. diag = (tmsg == ntmsg) ? "adjacently" : "after gap";
  231. goto mfound;
  232. }
  233. }
  234. for (tmsg = svars->ctx[t]->msgs; tmsg != ntmsg; tmsg = tmsg->next) {
  235. if (tmsg->status & M_DEAD)
  236. continue;
  237. if (tmsg->tuid[0] && !memcmp( tmsg->tuid, srec->tuid, TUIDL )) {
  238. diag = "after reset";
  239. goto mfound;
  240. }
  241. }
  242. debug( " -> TUID lost\n" );
  243. jFprintf( svars, "& %d %d\n", srec->uid[M], srec->uid[S] );
  244. srec->flags = 0;
  245. srec->tuid[0] = 0;
  246. num_lost++;
  247. continue;
  248. mfound:
  249. debug( " -> new UID %d %s\n", tmsg->uid, diag );
  250. jFprintf( svars, "%c %d %d %d\n", "<>"[t], srec->uid[M], srec->uid[S], tmsg->uid );
  251. tmsg->srec = srec;
  252. srec->msg[t] = tmsg;
  253. ntmsg = tmsg->next;
  254. srec->uid[t] = tmsg->uid;
  255. srec->tuid[0] = 0;
  256. }
  257. }
  258. if (num_lost)
  259. warn( "Warning: lost track of %d %sed message(s)\n", num_lost, str_hl[t] );
  260. }
  261. typedef struct copy_vars {
  262. void (*cb)( int sts, int uid, struct copy_vars *vars );
  263. void *aux;
  264. sync_rec_t *srec; /* also ->tuid */
  265. message_t *msg;
  266. msg_data_t data;
  267. } copy_vars_t;
  268. static void msg_fetched( int sts, void *aux );
  269. static void
  270. copy_msg( copy_vars_t *vars )
  271. {
  272. DECL_INIT_SVARS(vars->aux);
  273. t ^= 1;
  274. vars->data.flags = vars->msg->flags;
  275. vars->data.date = svars->chan->use_internal_date ? -1 : 0;
  276. svars->drv[t]->fetch_msg( svars->ctx[t], vars->msg, &vars->data, msg_fetched, vars );
  277. }
  278. static void msg_stored( int sts, int uid, void *aux );
  279. static void
  280. copy_msg_bytes( char **out_ptr, const char *in_buf, int *in_idx, int in_len, int in_cr, int out_cr )
  281. {
  282. char *out = *out_ptr;
  283. int idx = *in_idx;
  284. if (out_cr != in_cr) {
  285. char c;
  286. if (out_cr) {
  287. for (; idx < in_len; idx++) {
  288. if ((c = in_buf[idx]) != '\r') {
  289. if (c == '\n')
  290. *out++ = '\r';
  291. *out++ = c;
  292. }
  293. }
  294. } else {
  295. for (; idx < in_len; idx++) {
  296. if ((c = in_buf[idx]) != '\r')
  297. *out++ = c;
  298. }
  299. }
  300. } else {
  301. memcpy( out, in_buf + idx, in_len - idx );
  302. out += in_len - idx;
  303. idx = in_len;
  304. }
  305. *out_ptr = out;
  306. *in_idx = idx;
  307. }
  308. static int
  309. copy_msg_convert( int in_cr, int out_cr, copy_vars_t *vars )
  310. {
  311. char *in_buf = vars->data.data;
  312. int in_len = vars->data.len;
  313. int idx = 0, sbreak = 0, ebreak = 0;
  314. int lines = 0, hdr_crs = 0, bdy_crs = 0, app_cr = 0, extra = 0;
  315. if (vars->srec) {
  316. nloop: ;
  317. int start = idx;
  318. int line_crs = 0;
  319. while (idx < in_len) {
  320. char c = in_buf[idx++];
  321. if (c == '\r') {
  322. line_crs++;
  323. } else if (c == '\n') {
  324. if (starts_with_upper( in_buf + start, in_len - start, "X-TUID: ", 8 )) {
  325. extra = (sbreak = start) - (ebreak = idx);
  326. goto oke;
  327. }
  328. lines++;
  329. hdr_crs += line_crs;
  330. if (idx - line_crs - 1 == start) {
  331. sbreak = ebreak = start;
  332. goto oke;
  333. }
  334. goto nloop;
  335. }
  336. }
  337. /* invalid message */
  338. free( in_buf );
  339. return 0;
  340. oke:
  341. app_cr = out_cr && (!in_cr || hdr_crs);
  342. extra += 8 + TUIDL + app_cr + 1;
  343. }
  344. if (out_cr != in_cr) {
  345. for (; idx < in_len; idx++) {
  346. char c = in_buf[idx];
  347. if (c == '\r')
  348. bdy_crs++;
  349. else if (c == '\n')
  350. lines++;
  351. }
  352. extra -= hdr_crs + bdy_crs;
  353. if (out_cr)
  354. extra += lines;
  355. }
  356. vars->data.len = in_len + extra;
  357. char *out_buf = vars->data.data = nfmalloc( vars->data.len );
  358. idx = 0;
  359. if (vars->srec) {
  360. copy_msg_bytes( &out_buf, in_buf, &idx, sbreak, in_cr, out_cr );
  361. memcpy( out_buf, "X-TUID: ", 8 );
  362. out_buf += 8;
  363. memcpy( out_buf, vars->srec->tuid, TUIDL );
  364. out_buf += TUIDL;
  365. if (app_cr)
  366. *out_buf++ = '\r';
  367. *out_buf++ = '\n';
  368. idx = ebreak;
  369. }
  370. copy_msg_bytes( &out_buf, in_buf, &idx, in_len, in_cr, out_cr );
  371. free( in_buf );
  372. return 1;
  373. }
  374. static void
  375. msg_fetched( int sts, void *aux )
  376. {
  377. copy_vars_t *vars = (copy_vars_t *)aux;
  378. DECL_SVARS;
  379. int scr, tcr;
  380. switch (sts) {
  381. case DRV_OK:
  382. INIT_SVARS(vars->aux);
  383. if (check_cancel( svars )) {
  384. free( vars->data.data );
  385. vars->cb( SYNC_CANCELED, 0, vars );
  386. return;
  387. }
  388. vars->msg->flags = vars->data.flags;
  389. scr = (svars->drv[1-t]->flags / DRV_CRLF) & 1;
  390. tcr = (svars->drv[t]->flags / DRV_CRLF) & 1;
  391. if (vars->srec || scr != tcr) {
  392. if (!copy_msg_convert( scr, tcr, vars )) {
  393. warn( "Warning: message %d from %s has incomplete header.\n",
  394. vars->msg->uid, str_ms[1-t] );
  395. vars->cb( SYNC_NOGOOD, 0, vars );
  396. return;
  397. }
  398. }
  399. svars->drv[t]->store_msg( svars->ctx[t], &vars->data, !vars->srec, msg_stored, vars );
  400. break;
  401. case DRV_CANCELED:
  402. vars->cb( SYNC_CANCELED, 0, vars );
  403. break;
  404. case DRV_MSG_BAD:
  405. vars->cb( SYNC_NOGOOD, 0, vars );
  406. break;
  407. default:
  408. vars->cb( SYNC_FAIL, 0, vars );
  409. break;
  410. }
  411. }
  412. static void
  413. msg_stored( int sts, int uid, void *aux )
  414. {
  415. copy_vars_t *vars = (copy_vars_t *)aux;
  416. DECL_SVARS;
  417. switch (sts) {
  418. case DRV_OK:
  419. vars->cb( SYNC_OK, uid, vars );
  420. break;
  421. case DRV_CANCELED:
  422. vars->cb( SYNC_CANCELED, 0, vars );
  423. break;
  424. case DRV_MSG_BAD:
  425. INIT_SVARS(vars->aux);
  426. (void)svars;
  427. warn( "Warning: %s refuses to store message %d from %s.\n",
  428. str_ms[t], vars->msg->uid, str_ms[1-t] );
  429. vars->cb( SYNC_NOGOOD, 0, vars );
  430. break;
  431. default:
  432. vars->cb( SYNC_FAIL, 0, vars );
  433. break;
  434. }
  435. }
  436. static void sync_bail( sync_vars_t *svars );
  437. static void sync_bail2( sync_vars_t *svars );
  438. static void sync_bail3( sync_vars_t *svars );
  439. static void cancel_done( void *aux );
  440. static void
  441. cancel_sync( sync_vars_t *svars )
  442. {
  443. int t;
  444. for (t = 0; t < 2; t++) {
  445. int other_state = svars->state[1-t];
  446. if (svars->ret & SYNC_BAD(t)) {
  447. cancel_done( AUX );
  448. } else if (!(svars->state[t] & ST_SENT_CANCEL)) {
  449. /* ignore subsequent failures from in-flight commands */
  450. svars->state[t] |= ST_SENT_CANCEL;
  451. svars->drv[t]->cancel_cmds( svars->ctx[t], cancel_done, AUX );
  452. }
  453. if (other_state & ST_CANCELED)
  454. break;
  455. }
  456. }
  457. static void
  458. cancel_done( void *aux )
  459. {
  460. DECL_INIT_SVARS(aux);
  461. svars->state[t] |= ST_CANCELED;
  462. if (svars->state[1-t] & ST_CANCELED) {
  463. if (svars->nfp) {
  464. Fclose( svars->nfp, 0 );
  465. Fclose( svars->jfp, 0 );
  466. }
  467. sync_bail( svars );
  468. }
  469. }
  470. static void
  471. store_bad( void *aux )
  472. {
  473. DECL_INIT_SVARS(aux);
  474. svars->drv[t]->cancel_store( svars->ctx[t] );
  475. svars->ret |= SYNC_BAD(t);
  476. cancel_sync( svars );
  477. }
  478. static int
  479. check_cancel( sync_vars_t *svars )
  480. {
  481. return (svars->state[M] | svars->state[S]) & (ST_SENT_CANCEL | ST_CANCELED);
  482. }
  483. static int
  484. check_ret( int sts, void *aux )
  485. {
  486. DECL_SVARS;
  487. if (sts == DRV_CANCELED)
  488. return 1;
  489. INIT_SVARS(aux);
  490. if (sts == DRV_BOX_BAD) {
  491. svars->ret |= SYNC_FAIL;
  492. cancel_sync( svars );
  493. return 1;
  494. }
  495. return check_cancel( svars );
  496. }
  497. #define SVARS_CHECK_RET \
  498. DECL_SVARS; \
  499. if (check_ret( sts, aux )) \
  500. return; \
  501. INIT_SVARS(aux)
  502. #define SVARS_CHECK_RET_VARS(type) \
  503. type *vars = (type *)aux; \
  504. DECL_SVARS; \
  505. if (check_ret( sts, vars->aux )) { \
  506. free( vars ); \
  507. return; \
  508. } \
  509. INIT_SVARS(vars->aux)
  510. #define SVARS_CHECK_CANCEL_RET \
  511. DECL_SVARS; \
  512. if (sts == SYNC_CANCELED) { \
  513. free( vars ); \
  514. return; \
  515. } \
  516. INIT_SVARS(vars->aux)
  517. static char *
  518. clean_strdup( const char *s )
  519. {
  520. char *cs;
  521. int i;
  522. cs = nfstrdup( s );
  523. for (i = 0; cs[i]; i++)
  524. if (cs[i] == '/')
  525. cs[i] = '!';
  526. return cs;
  527. }
  528. #define JOURNAL_VERSION "3"
  529. static int
  530. prepare_state( sync_vars_t *svars )
  531. {
  532. char *s, *cmname, *csname;
  533. channel_conf_t *chan;
  534. chan = svars->chan;
  535. if (!strcmp( chan->sync_state ? chan->sync_state : global_conf.sync_state, "*" )) {
  536. if (!svars->ctx[S]->path) {
  537. error( "Error: store '%s' does not support in-box sync state\n", chan->stores[S]->name );
  538. return 0;
  539. }
  540. nfasprintf( &svars->dname, "%s/." EXE "state", svars->ctx[S]->path );
  541. } else {
  542. csname = clean_strdup( svars->box_name[S] );
  543. if (chan->sync_state)
  544. nfasprintf( &svars->dname, "%s%s", chan->sync_state, csname );
  545. else {
  546. char c = FieldDelimiter;
  547. cmname = clean_strdup( svars->box_name[M] );
  548. nfasprintf( &svars->dname, "%s%c%s%c%s_%c%s%c%s", global_conf.sync_state,
  549. c, chan->stores[M]->name, c, cmname, c, chan->stores[S]->name, c, csname );
  550. free( cmname );
  551. }
  552. free( csname );
  553. if (!(s = strrchr( svars->dname, '/' ))) {
  554. error( "Error: invalid SyncState location '%s'\n", svars->dname );
  555. return 0;
  556. }
  557. *s = 0;
  558. if (mkdir( svars->dname, 0700 ) && errno != EEXIST) {
  559. sys_error( "Error: cannot create SyncState directory '%s'", svars->dname );
  560. return 0;
  561. }
  562. *s = '/';
  563. }
  564. nfasprintf( &svars->jname, "%s.journal", svars->dname );
  565. nfasprintf( &svars->nname, "%s.new", svars->dname );
  566. nfasprintf( &svars->lname, "%s.lock", svars->dname );
  567. return 1;
  568. }
  569. static int
  570. lock_state( sync_vars_t *svars )
  571. {
  572. struct flock lck;
  573. if (svars->lfd >= 0)
  574. return 1;
  575. memset( &lck, 0, sizeof(lck) );
  576. #if SEEK_SET != 0
  577. lck.l_whence = SEEK_SET;
  578. #endif
  579. #if F_WRLCK != 0
  580. lck.l_type = F_WRLCK;
  581. #endif
  582. if ((svars->lfd = open( svars->lname, O_WRONLY|O_CREAT, 0666 )) < 0) {
  583. sys_error( "Error: cannot create lock file %s", svars->lname );
  584. return 0;
  585. }
  586. if (fcntl( svars->lfd, F_SETLK, &lck )) {
  587. error( "Error: channel :%s:%s-:%s:%s is locked\n",
  588. svars->chan->stores[M]->name, svars->orig_name[M], svars->chan->stores[S]->name, svars->orig_name[S] );
  589. close( svars->lfd );
  590. svars->lfd = -1;
  591. return 0;
  592. }
  593. return 1;
  594. }
  595. static void
  596. save_state( sync_vars_t *svars )
  597. {
  598. sync_rec_t *srec;
  599. char fbuf[16]; /* enlarge when support for keywords is added */
  600. Fprintf( svars->nfp,
  601. "MasterUidValidity %d\nSlaveUidValidity %d\nMaxPulledUid %d\nMaxPushedUid %d\n",
  602. svars->uidval[M], svars->uidval[S], svars->maxuid[M], svars->maxuid[S] );
  603. if (svars->smaxxuid)
  604. Fprintf( svars->nfp, "MaxExpiredSlaveUid %d\n", svars->smaxxuid );
  605. Fprintf( svars->nfp, "\n" );
  606. for (srec = svars->srecs; srec; srec = srec->next) {
  607. if (srec->status & S_DEAD)
  608. continue;
  609. make_flags( srec->flags, fbuf );
  610. Fprintf( svars->nfp, "%d %d %s%s\n", srec->uid[M], srec->uid[S],
  611. srec->status & S_EXPIRED ? "X" : "", fbuf );
  612. }
  613. Fclose( svars->nfp, 1 );
  614. Fclose( svars->jfp, 0 );
  615. if (!(DFlags & KEEPJOURNAL)) {
  616. /* order is important! */
  617. if (rename( svars->nname, svars->dname ))
  618. warn( "Warning: cannot commit sync state %s\n", svars->dname );
  619. else if (unlink( svars->jname ))
  620. warn( "Warning: cannot delete journal %s\n", svars->jname );
  621. }
  622. }
  623. static int
  624. load_state( sync_vars_t *svars )
  625. {
  626. sync_rec_t *srec, *nsrec;
  627. char *s;
  628. FILE *jfp;
  629. int line, t, t1, t2, t3;
  630. char c;
  631. struct stat st;
  632. char fbuf[16]; /* enlarge when support for keywords is added */
  633. char buf[128], buf1[64], buf2[64];
  634. if ((jfp = fopen( svars->dname, "r" ))) {
  635. if (!lock_state( svars ))
  636. goto jbail;
  637. debug( "reading sync state %s ...\n", svars->dname );
  638. line = 0;
  639. while (fgets( buf, sizeof(buf), jfp )) {
  640. line++;
  641. if (!(t = strlen( buf )) || buf[t - 1] != '\n') {
  642. error( "Error: incomplete sync state header entry at %s:%d\n", svars->dname, line );
  643. jbail:
  644. fclose( jfp );
  645. return 0;
  646. }
  647. if (t == 1)
  648. goto gothdr;
  649. if (line == 1 && isdigit( buf[0] )) {
  650. if (sscanf( buf, "%63s %63s", buf1, buf2 ) != 2 ||
  651. sscanf( buf1, "%d:%d", &svars->uidval[M], &svars->maxuid[M] ) < 2 ||
  652. sscanf( buf2, "%d:%d:%d", &svars->uidval[S], &svars->smaxxuid, &svars->maxuid[S] ) < 3) {
  653. error( "Error: invalid sync state header in %s\n", svars->dname );
  654. goto jbail;
  655. }
  656. goto gothdr;
  657. }
  658. if (sscanf( buf, "%63s %d", buf1, &t1 ) != 2) {
  659. error( "Error: malformed sync state header entry at %s:%d\n", svars->dname, line );
  660. goto jbail;
  661. }
  662. if (!strcmp( buf1, "MasterUidValidity" ))
  663. svars->uidval[M] = t1;
  664. else if (!strcmp( buf1, "SlaveUidValidity" ))
  665. svars->uidval[S] = t1;
  666. else if (!strcmp( buf1, "MaxPulledUid" ))
  667. svars->maxuid[M] = t1;
  668. else if (!strcmp( buf1, "MaxPushedUid" ))
  669. svars->maxuid[S] = t1;
  670. else if (!strcmp( buf1, "MaxExpiredSlaveUid" ))
  671. svars->smaxxuid = t1;
  672. else {
  673. error( "Error: unrecognized sync state header entry at %s:%d\n", svars->dname, line );
  674. goto jbail;
  675. }
  676. }
  677. error( "Error: unterminated sync state header in %s\n", svars->dname );
  678. goto jbail;
  679. gothdr:
  680. while (fgets( buf, sizeof(buf), jfp )) {
  681. line++;
  682. if (!(t = strlen( buf )) || buf[--t] != '\n') {
  683. error( "Error: incomplete sync state entry at %s:%d\n", svars->dname, line );
  684. goto jbail;
  685. }
  686. buf[t] = 0;
  687. fbuf[0] = 0;
  688. if (sscanf( buf, "%d %d %15s", &t1, &t2, fbuf ) < 2) {
  689. error( "Error: invalid sync state entry at %s:%d\n", svars->dname, line );
  690. goto jbail;
  691. }
  692. srec = nfmalloc( sizeof(*srec) );
  693. srec->uid[M] = t1;
  694. srec->uid[S] = t2;
  695. s = fbuf;
  696. if (*s == 'X') {
  697. s++;
  698. srec->status = S_EXPIRE | S_EXPIRED;
  699. } else
  700. srec->status = 0;
  701. srec->flags = parse_flags( s );
  702. debug( " entry (%d,%d,%u,%s)\n", srec->uid[M], srec->uid[S], srec->flags, srec->status & S_EXPIRED ? "X" : "" );
  703. srec->msg[M] = srec->msg[S] = 0;
  704. srec->tuid[0] = 0;
  705. srec->next = 0;
  706. *svars->srecadd = srec;
  707. svars->srecadd = &srec->next;
  708. svars->nsrecs++;
  709. }
  710. fclose( jfp );
  711. svars->existing = 1;
  712. } else {
  713. if (errno != ENOENT) {
  714. sys_error( "Error: cannot read sync state %s", svars->dname );
  715. return 0;
  716. }
  717. svars->existing = 0;
  718. }
  719. svars->newmaxuid[M] = svars->maxuid[M];
  720. svars->newmaxuid[S] = svars->maxuid[S];
  721. svars->mmaxxuid = INT_MAX;
  722. line = 0;
  723. if ((jfp = fopen( svars->jname, "r" ))) {
  724. if (!lock_state( svars ))
  725. goto jbail;
  726. if (!stat( svars->nname, &st ) && fgets( buf, sizeof(buf), jfp )) {
  727. debug( "recovering journal ...\n" );
  728. if (!(t = strlen( buf )) || buf[--t] != '\n') {
  729. error( "Error: incomplete journal header in %s\n", svars->jname );
  730. goto jbail;
  731. }
  732. buf[t] = 0;
  733. if (!equals( buf, t, JOURNAL_VERSION, strlen(JOURNAL_VERSION) )) {
  734. error( "Error: incompatible journal version "
  735. "(got %s, expected " JOURNAL_VERSION ")\n", buf );
  736. goto jbail;
  737. }
  738. srec = 0;
  739. line = 1;
  740. while (fgets( buf, sizeof(buf), jfp )) {
  741. line++;
  742. if (!(t = strlen( buf )) || buf[--t] != '\n') {
  743. error( "Error: incomplete journal entry at %s:%d\n", svars->jname, line );
  744. goto jbail;
  745. }
  746. buf[t] = 0;
  747. if ((c = buf[0]) == '#' ?
  748. (t3 = 0, (sscanf( buf + 2, "%d %d %n", &t1, &t2, &t3 ) < 2) || !t3 || (t - t3 != TUIDL + 2)) :
  749. c == 'S' || c == '!' ?
  750. (sscanf( buf + 2, "%d", &t1 ) != 1) :
  751. c == 'F' || c == 'T' || c == '+' || c == '&' || c == '-' || c == '|' || c == '/' || c == '\\' ?
  752. (sscanf( buf + 2, "%d %d", &t1, &t2 ) != 2) :
  753. (sscanf( buf + 2, "%d %d %d", &t1, &t2, &t3 ) != 3))
  754. {
  755. error( "Error: malformed journal entry at %s:%d\n", svars->jname, line );
  756. goto jbail;
  757. }
  758. if (c == 'S')
  759. svars->maxuid[t1] = svars->newmaxuid[t1];
  760. else if (c == 'F')
  761. svars->newuid[t1] = t2;
  762. else if (c == 'T')
  763. *int_array_append( &svars->trashed_msgs[t1] ) = t2;
  764. else if (c == '!')
  765. svars->smaxxuid = t1;
  766. else if (c == '|') {
  767. svars->uidval[M] = t1;
  768. svars->uidval[S] = t2;
  769. } else if (c == '+') {
  770. srec = nfmalloc( sizeof(*srec) );
  771. srec->uid[M] = t1;
  772. srec->uid[S] = t2;
  773. if (svars->newmaxuid[M] < t1)
  774. svars->newmaxuid[M] = t1;
  775. if (svars->newmaxuid[S] < t2)
  776. svars->newmaxuid[S] = t2;
  777. debug( " new entry(%d,%d)\n", t1, t2 );
  778. srec->msg[M] = srec->msg[S] = 0;
  779. srec->status = 0;
  780. srec->flags = 0;
  781. srec->tuid[0] = 0;
  782. srec->next = 0;
  783. *svars->srecadd = srec;
  784. svars->srecadd = &srec->next;
  785. svars->nsrecs++;
  786. } else {
  787. for (nsrec = srec; srec; srec = srec->next)
  788. if (srec->uid[M] == t1 && srec->uid[S] == t2)
  789. goto syncfnd;
  790. for (srec = svars->srecs; srec != nsrec; srec = srec->next)
  791. if (srec->uid[M] == t1 && srec->uid[S] == t2)
  792. goto syncfnd;
  793. error( "Error: journal entry at %s:%d refers to non-existing sync state entry\n", svars->jname, line );
  794. goto jbail;
  795. syncfnd:
  796. debugn( " entry(%d,%d,%u) ", srec->uid[M], srec->uid[S], srec->flags );
  797. switch (c) {
  798. case '-':
  799. debug( "killed\n" );
  800. if (srec->msg[M])
  801. srec->msg[M]->srec = 0;
  802. srec->status = S_DEAD;
  803. break;
  804. case '#':
  805. memcpy( srec->tuid, buf + t3 + 2, TUIDL );
  806. debug( "TUID now %." stringify(TUIDL) "s\n", srec->tuid );
  807. break;
  808. case '&':
  809. debug( "TUID %." stringify(TUIDL) "s lost\n", srec->tuid );
  810. srec->flags = 0;
  811. srec->tuid[0] = 0;
  812. break;
  813. case '<':
  814. debug( "master now %d\n", t3 );
  815. srec->uid[M] = t3;
  816. srec->tuid[0] = 0;
  817. break;
  818. case '>':
  819. debug( "slave now %d\n", t3 );
  820. srec->uid[S] = t3;
  821. srec->tuid[0] = 0;
  822. break;
  823. case '*':
  824. debug( "flags now %d\n", t3 );
  825. srec->flags = t3;
  826. break;
  827. case '~':
  828. debug( "expire now %d\n", t3 );
  829. if (t3)
  830. srec->status |= S_EXPIRE;
  831. else
  832. srec->status &= ~S_EXPIRE;
  833. break;
  834. case '\\':
  835. t3 = (srec->status & S_EXPIRED);
  836. debug( "expire back to %d\n", t3 / S_EXPIRED );
  837. if (t3)
  838. srec->status |= S_EXPIRE;
  839. else
  840. srec->status &= ~S_EXPIRE;
  841. break;
  842. case '/':
  843. t3 = (srec->status & S_EXPIRE);
  844. debug( "expired now %d\n", t3 / S_EXPIRE );
  845. if (t3) {
  846. if (svars->smaxxuid < srec->uid[S])
  847. svars->smaxxuid = srec->uid[S];
  848. srec->status |= S_EXPIRED;
  849. } else
  850. srec->status &= ~S_EXPIRED;
  851. break;
  852. default:
  853. error( "Error: unrecognized journal entry at %s:%d\n", svars->jname, line );
  854. goto jbail;
  855. }
  856. }
  857. }
  858. }
  859. fclose( jfp );
  860. } else {
  861. if (errno != ENOENT) {
  862. sys_error( "Error: cannot read journal %s", svars->jname );
  863. return 0;
  864. }
  865. }
  866. svars->replayed = line;
  867. return 1;
  868. }
  869. static void
  870. delete_state( sync_vars_t *svars )
  871. {
  872. unlink( svars->nname );
  873. unlink( svars->jname );
  874. if (unlink( svars->dname ) || unlink( svars->lname )) {
  875. sys_error( "Error: channel %s: sync state cannot be deleted", svars->chan->name );
  876. svars->ret = SYNC_FAIL;
  877. }
  878. }
  879. static void box_confirmed( int sts, void *aux );
  880. static void box_confirmed2( sync_vars_t *svars, int t );
  881. static void box_deleted( int sts, void *aux );
  882. static void box_created( int sts, void *aux );
  883. static void box_opened( int sts, void *aux );
  884. static void box_opened2( sync_vars_t *svars, int t );
  885. static void load_box( sync_vars_t *svars, int t, int minwuid, int_array_t mexcs );
  886. void
  887. sync_boxes( store_t *ctx[], const char *names[], int present[], channel_conf_t *chan,
  888. void (*cb)( int sts, void *aux ), void *aux )
  889. {
  890. sync_vars_t *svars;
  891. int t;
  892. svars = nfcalloc( sizeof(*svars) );
  893. svars->t[1] = 1;
  894. svars->ref_count = 1;
  895. svars->cb = cb;
  896. svars->aux = aux;
  897. svars->ctx[0] = ctx[0];
  898. svars->ctx[1] = ctx[1];
  899. svars->chan = chan;
  900. svars->lfd = -1;
  901. svars->uidval[0] = svars->uidval[1] = -1;
  902. svars->srecadd = &svars->srecs;
  903. for (t = 0; t < 2; t++) {
  904. svars->orig_name[t] =
  905. (!names[t] || (ctx[t]->conf->map_inbox && !strcmp( ctx[t]->conf->map_inbox, names[t] ))) ?
  906. "INBOX" : names[t];
  907. if (!ctx[t]->conf->flat_delim) {
  908. svars->box_name[t] = nfstrdup( svars->orig_name[t] );
  909. } else if (map_name( svars->orig_name[t], &svars->box_name[t], 0, "/", ctx[t]->conf->flat_delim ) < 0) {
  910. error( "Error: canonical mailbox name '%s' contains flattened hierarchy delimiter\n", svars->orig_name[t] );
  911. bail3:
  912. svars->ret = SYNC_FAIL;
  913. sync_bail3( svars );
  914. return;
  915. }
  916. ctx[t]->uidvalidity = -1;
  917. set_bad_callback( ctx[t], store_bad, AUX );
  918. svars->drv[t] = ctx[t]->conf->driver;
  919. }
  920. /* Both boxes must be fully set up at this point, so that error exit paths
  921. * don't run into uninitialized variables. */
  922. for (t = 0; t < 2; t++) {
  923. switch (svars->drv[t]->select_box( ctx[t], svars->box_name[t] )) {
  924. case DRV_CANCELED:
  925. store_bad( AUX );
  926. return;
  927. case DRV_BOX_BAD:
  928. goto bail3;
  929. }
  930. }
  931. if (!prepare_state( svars )) {
  932. svars->ret = SYNC_FAIL;
  933. sync_bail2( svars );
  934. return;
  935. }
  936. if (!load_state( svars )) {
  937. svars->ret = SYNC_FAIL;
  938. sync_bail( svars );
  939. return;
  940. }
  941. sync_ref( svars );
  942. for (t = 0; ; t++) {
  943. info( "Opening %s box %s...\n", str_ms[t], svars->orig_name[t] );
  944. if (present[t] == BOX_ABSENT)
  945. box_confirmed2( svars, t );
  946. else
  947. svars->drv[t]->open_box( ctx[t], box_confirmed, AUX );
  948. if (t || check_cancel( svars ))
  949. break;
  950. }
  951. sync_deref( svars );
  952. }
  953. static void
  954. box_confirmed( int sts, void *aux )
  955. {
  956. DECL_SVARS;
  957. if (sts == DRV_CANCELED)
  958. return;
  959. INIT_SVARS(aux);
  960. if (check_cancel( svars ))
  961. return;
  962. if (sts == DRV_OK)
  963. svars->state[t] |= ST_PRESENT;
  964. box_confirmed2( svars, t );
  965. }
  966. static void
  967. box_confirmed2( sync_vars_t *svars, int t )
  968. {
  969. svars->state[t] |= ST_CONFIRMED;
  970. if (!(svars->state[1-t] & ST_CONFIRMED))
  971. return;
  972. sync_ref( svars );
  973. for (t = 0; ; t++) {
  974. if (!(svars->state[t] & ST_PRESENT)) {
  975. if (!(svars->state[1-t] & ST_PRESENT)) {
  976. if (!svars->existing) {
  977. error( "Error: channel %s: both master %s and slave %s cannot be opened.\n",
  978. svars->chan->name, svars->orig_name[M], svars->orig_name[S] );
  979. bail:
  980. svars->ret = SYNC_FAIL;
  981. } else {
  982. /* This can legitimately happen if a deletion propagation was interrupted.
  983. * We have no place to record this transaction, so we just assume it.
  984. * Of course this bears the danger of clearing the state if both mailboxes
  985. * temorarily cannot be opened for some weird reason (while the stores can). */
  986. delete_state( svars );
  987. }
  988. done:
  989. sync_bail( svars );
  990. break;
  991. }
  992. if (svars->existing) {
  993. if (!(svars->chan->ops[1-t] & OP_REMOVE)) {
  994. error( "Error: channel %s: %s %s cannot be opened.\n",
  995. svars->chan->name, str_ms[t], svars->orig_name[t] );
  996. goto bail;
  997. }
  998. if (svars->drv[1-t]->confirm_box_empty( svars->ctx[1-t] ) != DRV_OK) {
  999. warn( "Warning: channel %s: %s %s cannot be opened and %s %s not empty.\n",
  1000. svars->chan->name, str_ms[t], svars->orig_name[t], str_ms[1-t], svars->orig_name[1-t] );
  1001. goto done;
  1002. }
  1003. info( "Deleting %s %s...\n", str_ms[1-t], svars->orig_name[1-t] );
  1004. svars->drv[1-t]->delete_box( svars->ctx[1-t], box_deleted, INV_AUX );
  1005. } else {
  1006. if (!(svars->chan->ops[t] & OP_CREATE)) {
  1007. box_opened( DRV_BOX_BAD, AUX );
  1008. } else {
  1009. info( "Creating %s %s...\n", str_ms[t], svars->orig_name[t] );
  1010. svars->drv[t]->create_box( svars->ctx[t], box_created, AUX );
  1011. }
  1012. }
  1013. } else {
  1014. box_opened2( svars, t );
  1015. }
  1016. if (t || check_cancel( svars ))
  1017. break;
  1018. }
  1019. sync_deref( svars );
  1020. }
  1021. static void
  1022. box_deleted( int sts, void *aux )
  1023. {
  1024. DECL_SVARS;
  1025. if (check_ret( sts, aux ))
  1026. return;
  1027. INIT_SVARS(aux);
  1028. delete_state( svars );
  1029. svars->drv[t]->finish_delete_box( svars->ctx[t] );
  1030. sync_bail( svars );
  1031. }
  1032. static void
  1033. box_created( int sts, void *aux )
  1034. {
  1035. DECL_SVARS;
  1036. if (check_ret( sts, aux ))
  1037. return;
  1038. INIT_SVARS(aux);
  1039. svars->drv[t]->open_box( svars->ctx[t], box_opened, AUX );
  1040. }
  1041. static void
  1042. box_opened( int sts, void *aux )
  1043. {
  1044. DECL_SVARS;
  1045. if (sts == DRV_CANCELED)
  1046. return;
  1047. INIT_SVARS(aux);
  1048. if (check_cancel( svars ))
  1049. return;
  1050. if (sts == DRV_BOX_BAD) {
  1051. error( "Error: channel %s: %s %s cannot be opened.\n",
  1052. svars->chan->name, str_ms[t], svars->orig_name[t] );
  1053. svars->ret = SYNC_FAIL;
  1054. sync_bail( svars );
  1055. } else {
  1056. box_opened2( svars, t );
  1057. }
  1058. }
  1059. static void
  1060. box_opened2( sync_vars_t *svars, int t )
  1061. {
  1062. store_t *ctx[2];
  1063. channel_conf_t *chan;
  1064. sync_rec_t *srec;
  1065. int_array_alloc_t mexcs;
  1066. int opts[2], fails, minwuid;
  1067. svars->state[t] |= ST_SELECTED;
  1068. if (!(svars->state[1-t] & ST_SELECTED))
  1069. return;
  1070. ctx[0] = svars->ctx[0];
  1071. ctx[1] = svars->ctx[1];
  1072. chan = svars->chan;
  1073. fails = 0;
  1074. for (t = 0; t < 2; t++)
  1075. if (svars->uidval[t] >= 0 && svars->uidval[t] != ctx[t]->uidvalidity)
  1076. fails++;
  1077. if (fails == 2) {
  1078. error( "Error: channel %s: UIDVALIDITY of both master and slave changed\n"
  1079. "(master got %d, expected %d; slave got %d, expected %d).\n",
  1080. svars->chan->name,
  1081. ctx[M]->uidvalidity, svars->uidval[M], ctx[S]->uidvalidity, svars->uidval[S] );
  1082. bail:
  1083. svars->ret = SYNC_FAIL;
  1084. sync_bail( svars );
  1085. return;
  1086. }
  1087. if (!lock_state( svars ))
  1088. goto bail;
  1089. if (!(svars->nfp = fopen( svars->nname, "w" ))) {
  1090. sys_error( "Error: cannot create new sync state %s", svars->nname );
  1091. goto bail;
  1092. }
  1093. if (!(svars->jfp = fopen( svars->jname, "a" ))) {
  1094. sys_error( "Error: cannot create journal %s", svars->jname );
  1095. fclose( svars->nfp );
  1096. goto bail;
  1097. }
  1098. setlinebuf( svars->jfp );
  1099. if (!svars->replayed)
  1100. jFprintf( svars, JOURNAL_VERSION "\n" );
  1101. opts[M] = opts[S] = 0;
  1102. if (fails)
  1103. opts[M] = opts[S] = OPEN_OLD|OPEN_OLD_IDS;
  1104. for (t = 0; t < 2; t++) {
  1105. if (chan->ops[t] & (OP_DELETE|OP_FLAGS)) {
  1106. opts[t] |= OPEN_SETFLAGS;
  1107. opts[1-t] |= OPEN_OLD;
  1108. if (chan->ops[t] & OP_FLAGS)
  1109. opts[1-t] |= OPEN_FLAGS;
  1110. }
  1111. if (chan->ops[t] & (OP_NEW|OP_RENEW)) {
  1112. opts[t] |= OPEN_APPEND;
  1113. if (chan->ops[t] & OP_RENEW)
  1114. opts[1-t] |= OPEN_OLD;
  1115. if (chan->ops[t] & OP_NEW)
  1116. opts[1-t] |= OPEN_NEW;
  1117. if (chan->ops[t] & OP_EXPUNGE)
  1118. opts[1-t] |= OPEN_FLAGS;
  1119. if (chan->stores[t]->max_size != INT_MAX) {
  1120. if (chan->ops[t] & OP_RENEW)
  1121. opts[1-t] |= OPEN_OLD_SIZE;
  1122. if (chan->ops[t] & OP_NEW)
  1123. opts[1-t] |= OPEN_NEW_SIZE;
  1124. }
  1125. }
  1126. if (chan->ops[t] & OP_EXPUNGE) {
  1127. opts[t] |= OPEN_EXPUNGE;
  1128. if (chan->stores[t]->trash) {
  1129. if (!chan->stores[t]->trash_only_new)
  1130. opts[t] |= OPEN_OLD;
  1131. opts[t] |= OPEN_NEW|OPEN_FLAGS;
  1132. } else if (chan->stores[1-t]->trash && chan->stores[1-t]->trash_remote_new)
  1133. opts[t] |= OPEN_NEW|OPEN_FLAGS;
  1134. }
  1135. }
  1136. if ((chan->ops[S] & (OP_NEW|OP_RENEW|OP_FLAGS)) && chan->max_messages)
  1137. opts[S] |= OPEN_OLD|OPEN_NEW|OPEN_FLAGS;
  1138. if (svars->replayed)
  1139. for (srec = svars->srecs; srec; srec = srec->next) {
  1140. if (srec->status & S_DEAD)
  1141. continue;
  1142. if (srec->tuid[0]) {
  1143. if (srec->uid[M] == -2)
  1144. opts[M] |= OPEN_NEW|OPEN_FIND, svars->state[M] |= ST_FIND_OLD;
  1145. else if (srec->uid[S] == -2)
  1146. opts[S] |= OPEN_NEW|OPEN_FIND, svars->state[S] |= ST_FIND_OLD;
  1147. else
  1148. assert( !"sync record with stray TUID" );
  1149. }
  1150. }
  1151. svars->drv[M]->prepare_load_box( ctx[M], opts[M] );
  1152. svars->drv[S]->prepare_load_box( ctx[S], opts[S] );
  1153. ARRAY_INIT( &mexcs );
  1154. if (svars->ctx[M]->opts & OPEN_OLD) {
  1155. if (chan->max_messages) {
  1156. /* When messages have been expired on the slave, the master fetch is split into
  1157. * two ranges: The bulk fetch which corresponds with the most recent messages, and an
  1158. * exception list of messages which would have been expired if they weren't important. */
  1159. debug( "preparing master selection - max expired slave uid is %d\n", svars->smaxxuid );
  1160. /* First, find out the lower bound for the bulk fetch. */
  1161. minwuid = INT_MAX;
  1162. for (srec = svars->srecs; srec; srec = srec->next) {
  1163. if ((srec->status & S_DEAD) || srec->uid[M] <= 0)
  1164. continue;
  1165. if (srec->status & S_EXPIRED) {
  1166. if (!srec->uid[S]) {
  1167. /* The expired message was already gone. */
  1168. continue;
  1169. }
  1170. /* The expired message was not expunged yet, so re-examine it.
  1171. * This will happen en masse, so just extend the bulk fetch. */
  1172. } else {
  1173. if (svars->smaxxuid >= srec->uid[S]) {
  1174. /* The non-expired message is in the generally expired range, so don't
  1175. * make it contribute to the bulk fetch. */
  1176. continue;
  1177. }
  1178. /* Usual non-expired message. */
  1179. }
  1180. if (minwuid > srec->uid[M])
  1181. minwuid = srec->uid[M];
  1182. }
  1183. debug( " min non-orphaned master uid is %d\n", minwuid );
  1184. /* Next, calculate the exception fetch. */
  1185. for (srec = svars->srecs; srec; srec = srec->next) {
  1186. if (srec->status & S_DEAD)
  1187. continue;
  1188. if (srec->uid[M] > 0 && srec->uid[S] > 0 && minwuid > srec->uid[M] &&
  1189. (!(svars->ctx[M]->opts & OPEN_NEW) || svars->maxuid[M] >= srec->uid[M])) {
  1190. /* The pair is alive, but outside the bulk range. */
  1191. *int_array_append( &mexcs ) = srec->uid[M];
  1192. }
  1193. }
  1194. debugn( " exception list is:" );
  1195. for (t = 0; t < mexcs.array.size; t++)
  1196. debugn( " %d", mexcs.array.data[t] );
  1197. debug( "\n" );
  1198. } else {
  1199. minwuid = 1;
  1200. }
  1201. } else {
  1202. minwuid = INT_MAX;
  1203. }
  1204. sync_ref( svars );
  1205. sort_int_array( mexcs.array );
  1206. load_box( svars, M, minwuid, mexcs.array );
  1207. if (!check_cancel( svars ))
  1208. load_box( svars, S, (ctx[S]->opts & OPEN_OLD) ? 1 : INT_MAX, (int_array_t){ 0, 0 } );
  1209. sync_deref( svars );
  1210. }
  1211. static int
  1212. get_seenuid( sync_vars_t *svars, int t )
  1213. {
  1214. int seenuid = 0;
  1215. for (sync_rec_t *srec = svars->srecs; srec; srec = srec->next)
  1216. if (!(srec->status & S_DEAD) && seenuid < srec->uid[t])
  1217. seenuid = srec->uid[t];
  1218. return seenuid;
  1219. }
  1220. static void box_loaded( int sts, void *aux );
  1221. static void
  1222. load_box( sync_vars_t *svars, int t, int minwuid, int_array_t mexcs )
  1223. {
  1224. int maxwuid, seenuid;
  1225. if (svars->ctx[t]->opts & OPEN_NEW) {
  1226. if (minwuid > svars->maxuid[t] + 1)
  1227. minwuid = svars->maxuid[t] + 1;
  1228. maxwuid = INT_MAX;
  1229. if (svars->ctx[t]->opts & (OPEN_OLD_IDS|OPEN_OLD_SIZE))
  1230. seenuid = get_seenuid( svars, t );
  1231. else
  1232. seenuid = 0;
  1233. } else if (svars->ctx[t]->opts & OPEN_OLD) {
  1234. maxwuid = seenuid = get_seenuid( svars, t );
  1235. } else
  1236. maxwuid = seenuid = 0;
  1237. if (seenuid < svars->maxuid[t]) {
  1238. /* We cannot rely on the maxuid, as uni-directional syncing does not update it.
  1239. * But if it is there, use it to avoid a possible gap in the fetched range. */
  1240. seenuid = svars->maxuid[t];
  1241. }
  1242. info( "Loading %s...\n", str_ms[t] );
  1243. if (maxwuid == INT_MAX)
  1244. debug( "loading %s [%d,inf] (new >= %d, seen <= %d)\n", str_ms[t], minwuid, svars->newuid[t], seenuid );
  1245. else
  1246. debug( "loading %s [%d,%d] (new >= %d, seen <= %d)\n", str_ms[t], minwuid, maxwuid, svars->newuid[t], seenuid );
  1247. svars->drv[t]->load_box( svars->ctx[t], minwuid, maxwuid, svars->newuid[t], seenuid, mexcs, box_loaded, AUX );
  1248. }
  1249. typedef struct {
  1250. void *aux;
  1251. sync_rec_t *srec;
  1252. int aflags, dflags;
  1253. } flag_vars_t;
  1254. typedef struct {
  1255. int uid;
  1256. sync_rec_t *srec;
  1257. } sync_rec_map_t;
  1258. static void flags_set( int sts, void *aux );
  1259. static void flags_set_p2( sync_vars_t *svars, sync_rec_t *srec, int t );
  1260. static void msgs_flags_set( sync_vars_t *svars, int t );
  1261. static void msg_copied( int sts, int uid, copy_vars_t *vars );
  1262. static void msg_copied_p2( sync_vars_t *svars, sync_rec_t *srec, int t, int uid );
  1263. static void msgs_copied( sync_vars_t *svars, int t );
  1264. static void
  1265. box_loaded( int sts, void *aux )
  1266. {
  1267. DECL_SVARS;
  1268. sync_rec_t *srec;
  1269. sync_rec_map_t *srecmap;
  1270. message_t *tmsg;
  1271. flag_vars_t *fv;
  1272. int uid, no[2], del[2], alive, todel, t1, t2;
  1273. int sflags, nflags, aflags, dflags;
  1274. uint hashsz, idx;
  1275. char fbuf[16]; /* enlarge when support for keywords is added */
  1276. if (check_ret( sts, aux ))
  1277. return;
  1278. INIT_SVARS(aux);
  1279. svars->state[t] |= ST_LOADED;
  1280. info( "%s: %d messages, %d recent\n", str_ms[t], svars->ctx[t]->count, svars->ctx[t]->recent );
  1281. if (svars->state[t] & ST_FIND_OLD) {
  1282. debug( "matching previously copied messages on %s\n", str_ms[t] );
  1283. match_tuids( svars, t );
  1284. }
  1285. debug( "matching messages on %s against sync records\n", str_ms[t] );
  1286. hashsz = bucketsForSize( svars->nsrecs * 3 );
  1287. srecmap = nfcalloc( hashsz * sizeof(*srecmap) );
  1288. for (srec = svars->srecs; srec; srec = srec->next) {
  1289. if (srec->status & S_DEAD)
  1290. continue;
  1291. uid = srec->uid[t];
  1292. if (uid <= 0)
  1293. continue;
  1294. idx = (uint)((uint)uid * 1103515245U) % hashsz;
  1295. while (srecmap[idx].uid)
  1296. if (++idx == hashsz)
  1297. idx = 0;
  1298. srecmap[idx].uid = uid;
  1299. srecmap[idx].srec = srec;
  1300. }
  1301. for (tmsg = svars->ctx[t]->msgs; tmsg; tmsg = tmsg->next) {
  1302. if (tmsg->srec) /* found by TUID */
  1303. continue;
  1304. uid = tmsg->uid;
  1305. if (DFlags & DEBUG_SYNC) {
  1306. make_flags( tmsg->flags, fbuf );
  1307. printf( tmsg->size ? " message %5d, %-4s, %6d: " : " message %5d, %-4s: ", uid, fbuf, tmsg->size );
  1308. }
  1309. idx = (uint)((uint)uid * 1103515245U) % hashsz;
  1310. while (srecmap[idx].uid) {
  1311. if (srecmap[idx].uid == uid) {
  1312. srec = srecmap[idx].srec;
  1313. goto found;
  1314. }
  1315. if (++idx == hashsz)
  1316. idx = 0;
  1317. }
  1318. debug( "new\n" );
  1319. continue;
  1320. found:
  1321. tmsg->srec = srec;
  1322. srec->msg[t] = tmsg;
  1323. debug( "pairs %5d\n", srec->uid[1-t] );
  1324. }
  1325. free( srecmap );
  1326. if (!(svars->state[1-t] & ST_LOADED))
  1327. return;
  1328. for (t = 0; t < 2; t++) {
  1329. if (svars->uidval[t] >= 0 && svars->uidval[t] != svars->ctx[t]->uidvalidity) {
  1330. unsigned need = 0, got = 0;
  1331. debug( "trying to re-approve uid validity of %s\n", str_ms[t] );
  1332. for (srec = svars->srecs; srec; srec = srec->next) {
  1333. if (srec->status & S_DEAD)
  1334. continue;
  1335. if (!srec->msg[t])
  1336. continue; // Message disappeared.
  1337. need++; // Present paired messages require re-validation.
  1338. if (!srec->msg[t]->msgid)
  1339. continue; // Messages without ID are useless for re-validation.
  1340. if (!srec->msg[1-t])
  1341. continue; // Partner disappeared.
  1342. if (!srec->msg[1-t]->msgid || strcmp( srec->msg[M]->msgid, srec->msg[S]->msgid )) {
  1343. error( "Error: channel %s, %s %s: UIDVALIDITY genuinely changed (at UID %d).\n",
  1344. svars->chan->name, str_ms[t], svars->orig_name[t], srec->uid[t] );
  1345. uvchg:
  1346. svars->ret |= SYNC_FAIL;
  1347. cancel_sync( svars );
  1348. return;
  1349. }
  1350. got++;
  1351. }
  1352. if (got < 20 && got * 5 < need * 4) {
  1353. // Too few confirmed messages. This is very likely in the drafts folder.
  1354. // A proper fallback would be fetching more headers (which potentially need
  1355. // normalization) or the message body (which should be truncated for sanity)
  1356. // and comparing.
  1357. error( "Error: channel %s, %s %s: Unable to recover from UIDVALIDITY change\n"
  1358. "(got %d, expected %d).\n",
  1359. svars->chan->name, str_ms[t], svars->orig_name[t],
  1360. svars->ctx[t]->uidvalidity, svars->uidval[t] );
  1361. goto uvchg;
  1362. }
  1363. notice( "Notice: channel %s, %s %s: Recovered from change of UIDVALIDITY.\n",
  1364. svars->chan->name, str_ms[t], svars->orig_name[t] );
  1365. svars->uidval[t] = -1;
  1366. }
  1367. }
  1368. if (svars->uidval[M] < 0 || svars->uidval[S] < 0) {
  1369. svars->uidval[M] = svars->ctx[M]->uidvalidity;
  1370. svars->uidval[S] = svars->ctx[S]->uidvalidity;
  1371. jFprintf( svars, "| %d %d\n", svars->uidval[M], svars->uidval[S] );
  1372. }
  1373. info( "Synchronizing...\n" );
  1374. debug( "synchronizing old entries\n" );
  1375. for (srec = svars->srecs; srec; srec = srec->next) {
  1376. if (srec->status & S_DEAD)
  1377. continue;
  1378. debug( "pair (%d,%d)\n", srec->uid[M], srec->uid[S] );
  1379. no[M] = !srec->msg[M] && (svars->ctx[M]->opts & OPEN_OLD);
  1380. no[S] = !srec->msg[S] && (svars->ctx[S]->opts & OPEN_OLD);
  1381. if (no[M] && no[S]) {
  1382. debug( " vanished\n" );
  1383. /* d.1) d.5) d.6) d.10) d.11) */
  1384. srec->status = S_DEAD;
  1385. jFprintf( svars, "- %d %d\n", srec->uid[M], srec->uid[S] );
  1386. } else {
  1387. del[M] = no[M] && (srec->uid[M] > 0);
  1388. del[S] = no[S] && (srec->uid[S] > 0);
  1389. for (t = 0; t < 2; t++) {
  1390. srec->aflags[t] = srec->dflags[t] = 0;
  1391. if (srec->msg[t] && (srec->msg[t]->flags & F_DELETED))
  1392. srec->status |= S_DEL(t);
  1393. /* excludes (push) c.3) d.2) d.3) d.4) / (pull) b.3) d.7) d.8) d.9) */
  1394. if (!srec->uid[t]) {
  1395. /* b.1) / c.1) */
  1396. debug( " no more %s\n", str_ms[t] );
  1397. } else if (del[1-t]) {
  1398. /* c.4) d.9) / b.4) d.4) */
  1399. if ((t == M) && (srec->status & (S_EXPIRE|S_EXPIRED))) {
  1400. /* Don't propagate deletion resulting from expiration. */
  1401. debug( " slave expired, orphaning master\n" );
  1402. jFprintf( svars, "> %d %d 0\n", srec->uid[M], srec->uid[S] );
  1403. srec->uid[S] = 0;
  1404. } else {
  1405. if (srec->msg[t] && (srec->msg[t]->status & M_FLAGS) && srec->msg[t]->flags != srec->flags)
  1406. notice( "Notice: conflicting changes in (%d,%d)\n", srec->uid[M], srec->uid[S] );
  1407. if (svars->chan->ops[t] & OP_DELETE) {
  1408. debug( " %sing delete\n", str_hl[t] );
  1409. srec->aflags[t] = F_DELETED;
  1410. srec->status |= S_DELETE;
  1411. } else {
  1412. debug( " not %sing delete\n", str_hl[t] );
  1413. }
  1414. }
  1415. } else if (!srec->msg[1-t])
  1416. /* c.1) c.2) d.7) d.8) / b.1) b.2) d.2) d.3) */
  1417. ;
  1418. else if (srec->uid[t] < 0)
  1419. /* b.2) / c.2) */
  1420. ; /* handled as new messages (sort of) */
  1421. else if (!del[t]) {
  1422. /* a) & b.3) / c.3) */
  1423. if (svars->chan->ops[t] & OP_FLAGS) {
  1424. sflags = srec->msg[1-t]->flags;
  1425. if ((t == M) && (srec->status & (S_EXPIRE|S_EXPIRED))) {
  1426. /* Don't propagate deletion resulting from expiration. */
  1427. debug( " slave expiring\n" );
  1428. sflags &= ~F_DELETED;
  1429. }
  1430. srec->aflags[t] = sflags & ~srec->flags;
  1431. srec->dflags[t] = ~sflags & srec->flags;
  1432. if (DFlags & DEBUG_SYNC) {
  1433. char afbuf[16], dfbuf[16]; /* enlarge when support for keywords is added */
  1434. make_flags( srec->aflags[t], afbuf );
  1435. make_flags( srec->dflags[t], dfbuf );
  1436. debug( " %sing flags: +%s -%s\n", str_hl[t], afbuf, dfbuf );
  1437. }
  1438. } else
  1439. debug( " not %sing flags\n", str_hl[t] );
  1440. } /* else b.4) / c.4) */
  1441. }
  1442. }
  1443. }
  1444. debug( "synchronizing new entries\n" );
  1445. for (t = 0; t < 2; t++) {
  1446. for (tmsg = svars->ctx[1-t]->msgs; tmsg; tmsg = tmsg->next) {
  1447. // If new have no srec, the message is always New. If we have a srec:
  1448. // - message is old (> 0) or expired (0) => ignore
  1449. // - message was skipped (-1) => ReNew
  1450. // - message was attempted, but failed (-2) => New
  1451. //
  1452. // If messages were previously ignored due to being excessive, they would now
  1453. // appear to be newer than the messages that got actually synced, so increment
  1454. // newmaxuid immediately to make sure we always look only at the newest ones.
  1455. // However, committing it to maxuid must be delayed until all messages were
  1456. // propagated, to ensure that all pending messages are still loaded next time
  1457. // in case of interruption - in particular skipping big messages would otherwise
  1458. // up the limit too early.
  1459. srec = tmsg->srec;
  1460. if (srec ? srec->uid[t] < 0 && (svars->chan->ops[t] & (srec->uid[t] == -1 ? OP_RENEW : OP_NEW))
  1461. : svars->newmaxuid[1-t] < tmsg->uid && (svars->chan->ops[t] & OP_NEW)) {
  1462. debug( "new message %d on %s\n", tmsg->uid, str_ms[1-t] );
  1463. if ((svars->chan->ops[t] & OP_EXPUNGE) && (tmsg->flags & F_DELETED)) {
  1464. debug( " -> not %sing - would be expunged anyway\n", str_hl[t] );
  1465. } else {
  1466. if (srec) {
  1467. debug( " -> pair(%d,%d) exists\n", srec->uid[M], srec->uid[S] );
  1468. } else {
  1469. srec = nfmalloc( sizeof(*srec) );
  1470. srec->next = 0;
  1471. *svars->srecadd = srec;
  1472. svars->srecadd = &srec->next;
  1473. svars->nsrecs++;
  1474. srec->status = 0;
  1475. srec->flags = 0;
  1476. srec->tuid[0] = 0;
  1477. srec->uid[1-t] = tmsg->uid;
  1478. srec->uid[t] = -2;
  1479. srec->msg[1-t] = tmsg;
  1480. srec->msg[t] = 0;
  1481. tmsg->srec = srec;
  1482. if (svars->newmaxuid[1-t] < tmsg->uid)
  1483. svars->newmaxuid[1-t] = tmsg->uid;
  1484. jFprintf( svars, "+ %d %d\n", srec->uid[M], srec->uid[S] );
  1485. debug( " -> pair(%d,%d) created\n", srec->uid[M], srec->uid[S] );
  1486. }
  1487. if ((tmsg->flags & F_FLAGGED) || tmsg->size <= svars->chan->stores[t]->max_size) {
  1488. if (tmsg->flags) {
  1489. srec->flags = tmsg->flags;
  1490. jFprintf( svars, "* %d %d %u\n", srec->uid[M], srec->uid[S], srec->flags );
  1491. debug( " -> updated flags to %u\n", tmsg->flags );
  1492. }
  1493. for (t1 = 0; t1 < TUIDL; t1++) {
  1494. t2 = arc4_getbyte() & 0x3f;
  1495. srec->tuid[t1] = t2 < 26 ? t2 + 'A' : t2 < 52 ? t2 + 'a' - 26 : t2 < 62 ? t2 + '0' - 52 : t2 == 62 ? '+' : '/';
  1496. }
  1497. jFprintf( svars, "# %d %d %." stringify(TUIDL) "s\n", srec->uid[M], srec->uid[S], srec->tuid );
  1498. debug( " -> %sing message, TUID %." stringify(TUIDL) "s\n", str_hl[t], srec->tuid );
  1499. } else {
  1500. if (srec->uid[t] == -1) {
  1501. debug( " -> not %sing - still too big\n", str_hl[t] );
  1502. } else {
  1503. debug( " -> not %sing - too big\n", str_hl[t] );
  1504. msg_copied_p2( svars, srec, t, -1 );
  1505. }
  1506. }
  1507. }
  1508. }
  1509. }
  1510. }
  1511. if ((svars->chan->ops[S] & (OP_NEW|OP_RENEW|OP_FLAGS)) && svars->chan->max_messages) {
  1512. /* Note: When this branch is entered, we have loaded all slave messages. */
  1513. /* Expire excess messages. Important (flagged, unread, or unpropagated) messages
  1514. * older than the first not expired message are not counted towards the total. */
  1515. debug( "preparing message expiration\n" );
  1516. alive = 0;
  1517. for (tmsg = svars->ctx[S]->msgs; tmsg; tmsg = tmsg->next) {
  1518. if (tmsg->status & M_DEAD)
  1519. continue;
  1520. if ((srec = tmsg->srec) && srec->uid[M] > 0 &&
  1521. ((tmsg->flags | srec->aflags[S]) & ~srec->dflags[S] & F_DELETED) &&
  1522. !(srec->status & (S_EXPIRE|S_EXPIRED))) {
  1523. /* Message was not propagated yet, or is deleted. */
  1524. } else {
  1525. alive++;
  1526. }
  1527. }
  1528. for (tmsg = svars->ctx[M]->msgs; tmsg; tmsg = tmsg->next) {
  1529. if ((srec = tmsg->srec) && srec->tuid[0] && !(tmsg->flags & F_DELETED))
  1530. alive++;
  1531. }
  1532. todel = alive - svars->chan->max_messages;
  1533. debug( "%d alive messages, %d excess - expiring\n", alive, todel );
  1534. alive = 0;
  1535. for (tmsg = svars->ctx[S]->msgs; tmsg; tmsg = tmsg->next) {
  1536. if (tmsg->status & M_DEAD)
  1537. continue;
  1538. if (!(srec = tmsg->srec) || srec->uid[M] <= 0) {
  1539. /* We did not push the message, so it must be kept. */
  1540. debug( " message %d unpropagated\n", tmsg->uid );
  1541. todel--;
  1542. } else {
  1543. nflags = (tmsg->flags | srec->aflags[S]) & ~srec->dflags[S];
  1544. if (!(nflags & F_DELETED) || (srec->status & (S_EXPIRE|S_EXPIRED))) {
  1545. /* The message is not deleted, or is already (being) expired. */
  1546. if ((nflags & F_FLAGGED) || !((nflags & F_SEEN) || ((void)(todel > 0 && alive++), svars->chan->expire_unread > 0))) {
  1547. /* Important messages are always kept. */
  1548. debug( " old pair(%d,%d) important\n", srec->uid[M], srec->uid[S] );
  1549. todel--;
  1550. } else if (todel > 0 ||
  1551. ((srec->status & (S_EXPIRE|S_EXPIRED)) == (S_EXPIRE|S_EXPIRED)) ||
  1552. ((srec->status & (S_EXPIRE|S_EXPIRED)) && (tmsg->flags & F_DELETED))) {
  1553. /* The message is excess or was already (being) expired. */
  1554. srec->status |= S_NEXPIRE;
  1555. debug( " old pair(%d,%d) expired\n", srec->uid[M], srec->uid[S] );
  1556. todel--;
  1557. }
  1558. }
  1559. }
  1560. }
  1561. for (tmsg = svars->ctx[M]->msgs; tmsg; tmsg = tmsg->next) {
  1562. if ((srec = tmsg->srec) && srec->tuid[0]) {
  1563. nflags = tmsg->flags;
  1564. if (!(nflags & F_DELETED)) {
  1565. if ((nflags & F_FLAGGED) || !((nflags & F_SEEN) || ((void)(todel > 0 && alive++), svars->chan->expire_unread > 0))) {
  1566. /* Important messages are always fetched. */
  1567. debug( " new pair(%d,%d) important\n", srec->uid[M], srec->uid[S] );
  1568. todel--;
  1569. } else if (todel > 0) {
  1570. /* The message is excess. */
  1571. srec->status |= S_NEXPIRE;
  1572. debug( " new pair(%d,%d) expired\n", srec->uid[M], srec->uid[S] );
  1573. svars->mmaxxuid = srec->uid[M];
  1574. todel--;
  1575. }
  1576. }
  1577. }
  1578. }
  1579. debug( "%d excess messages remain\n", todel );
  1580. if (svars->chan->expire_unread < 0 && (uint)alive * 2 > svars->chan->max_messages) {
  1581. error( "%s: %d unread messages in excess of MaxMessages (%d).\n"
  1582. "Please set ExpireUnread to decide outcome. Skipping mailbox.\n",
  1583. svars->orig_name[S], alive, svars->chan->max_messages );
  1584. svars->ret |= SYNC_FAIL;
  1585. cancel_sync( svars );
  1586. return;
  1587. }
  1588. for (srec = svars->srecs; srec; srec = srec->next) {
  1589. if (srec->status & S_DEAD)
  1590. continue;
  1591. if (!srec->tuid[0]) {
  1592. if (!srec->msg[S])
  1593. continue;
  1594. uint nex = (srec->status / S_NEXPIRE) & 1;
  1595. if (nex != ((srec->status / S_EXPIRED) & 1)) {
  1596. /* The record needs a state change ... */
  1597. if (nex != ((srec->status / S_EXPIRE) & 1)) {
  1598. /* ... and we need to start a transaction. */
  1599. jFprintf( svars, "~ %d %d %d\n", srec->uid[M], srec->uid[S], nex );
  1600. debug( " pair(%d,%d): %d (pre)\n", srec->uid[M], srec->uid[S], nex );
  1601. srec->status = (srec->status & ~S_EXPIRE) | (nex * S_EXPIRE);
  1602. } else {
  1603. /* ... but the "right" transaction is already pending. */
  1604. debug( " pair(%d,%d): %d (pending)\n", srec->uid[M], srec->uid[S], nex );
  1605. }
  1606. } else {
  1607. /* Note: the "wrong" transaction may be pending here,
  1608. * e.g.: S_NEXPIRE = 0, S_EXPIRE = 1, S_EXPIRED = 0. */
  1609. }
  1610. } else {
  1611. if (srec->status & S_NEXPIRE) {
  1612. jFprintf( svars, "- %d %d\n", srec->uid[M], srec->uid[S] );
  1613. debug( " pair(%d,%d): 1 (abort)\n", srec->uid[M], srec->uid[S] );
  1614. srec->msg[M]->srec = 0;
  1615. srec->status = S_DEAD;
  1616. }
  1617. }
  1618. }
  1619. }
  1620. sync_ref( svars );
  1621. debug( "synchronizing flags\n" );
  1622. for (srec = svars->srecs; srec; srec = srec->next) {
  1623. if ((srec->status & S_DEAD) || srec->uid[M] <= 0 || srec->uid[S] <= 0)
  1624. continue;
  1625. for (t = 0; t < 2; t++) {
  1626. aflags = srec->aflags[t];
  1627. dflags = srec->dflags[t];
  1628. if (srec->status & S_DELETE) {
  1629. if (!aflags) {
  1630. /* This deletion propagation goes the other way round. */
  1631. continue;
  1632. }
  1633. } else {
  1634. /* The trigger is an expiration transaction being ongoing ... */
  1635. if ((t == S) && ((shifted_bit(srec->status, S_EXPIRE, S_EXPIRED) ^ srec->status) & S_EXPIRED)) {
  1636. /* ... but the actual action derives from the wanted state. */
  1637. if (srec->status & S_NEXPIRE)
  1638. aflags |= F_DELETED;
  1639. else
  1640. dflags |= F_DELETED;
  1641. }
  1642. }
  1643. if ((svars->chan->ops[t] & OP_EXPUNGE) && (((srec->msg[t] ? srec->msg[t]->flags : 0) | aflags) & ~dflags & F_DELETED) &&
  1644. (!svars->ctx[t]->conf->trash || svars->ctx[t]->conf->trash_only_new))
  1645. {
  1646. /* If the message is going to be expunged, don't propagate anything but the deletion. */
  1647. srec->aflags[t] &= F_DELETED;
  1648. aflags &= F_DELETED;
  1649. srec->dflags[t] = dflags = 0;
  1650. }
  1651. if (srec->msg[t] && (srec->msg[t]->status & M_FLAGS)) {
  1652. /* If we know the target message's state, optimize away non-changes. */
  1653. aflags &= ~srec->msg[t]->flags;
  1654. dflags &= srec->msg[t]->flags;
  1655. }
  1656. if (aflags | dflags) {
  1657. flags_total[t]++;
  1658. stats();
  1659. svars->flags_pending[t]++;
  1660. fv = nfmalloc( sizeof(*fv) );
  1661. fv->aux = AUX;
  1662. fv->srec = srec;
  1663. fv->aflags = aflags;
  1664. fv->dflags = dflags;
  1665. svars->drv[t]->set_msg_flags( svars->ctx[t], srec->msg[t], srec->uid[t], aflags, dflags, flags_set, fv );
  1666. if (check_cancel( svars ))
  1667. goto out;
  1668. } else
  1669. flags_set_p2( svars, srec, t );
  1670. }
  1671. }
  1672. for (t = 0; t < 2; t++) {
  1673. svars->drv[t]->commit_cmds( svars->ctx[t] );
  1674. svars->state[t] |= ST_SENT_FLAGS;
  1675. msgs_flags_set( svars, t );
  1676. if (check_cancel( svars ))
  1677. goto out;
  1678. }
  1679. debug( "propagating new messages\n" );
  1680. if (UseFSync)
  1681. fdatasync( fileno( svars->jfp ) );
  1682. for (t = 0; t < 2; t++) {
  1683. svars->newuid[t] = svars->ctx[t]->uidnext;
  1684. jFprintf( svars, "F %d %d\n", t, svars->newuid[t] );
  1685. svars->new_msgs[t] = svars->ctx[1-t]->msgs;
  1686. msgs_copied( svars, t );
  1687. if (check_cancel( svars ))
  1688. goto out;
  1689. }
  1690. out:
  1691. sync_deref( svars );
  1692. }
  1693. static void
  1694. msg_copied( int sts, int uid, copy_vars_t *vars )
  1695. {
  1696. SVARS_CHECK_CANCEL_RET;
  1697. switch (sts) {
  1698. case SYNC_OK:
  1699. if (uid < 0)
  1700. svars->state[t] |= ST_FIND_NEW;
  1701. msg_copied_p2( svars, vars->srec, t, uid );
  1702. break;
  1703. case SYNC_NOGOOD:
  1704. debug( " -> killing (%d,%d)\n", vars->srec->uid[M], vars->srec->uid[S] );
  1705. vars->srec->status = S_DEAD;
  1706. jFprintf( svars, "- %d %d\n", vars->srec->uid[M], vars->srec->uid[S] );
  1707. break;
  1708. default:
  1709. cancel_sync( svars );
  1710. free( vars );
  1711. return;
  1712. }
  1713. free( vars );
  1714. new_done[t]++;
  1715. stats();
  1716. svars->new_pending[t]--;
  1717. msgs_copied( svars, t );
  1718. }
  1719. static void
  1720. msg_copied_p2( sync_vars_t *svars, sync_rec_t *srec, int t, int uid )
  1721. {
  1722. /* Possible previous UIDs:
  1723. * - -2 when the entry is new
  1724. * - -1 when re-newing an entry
  1725. * Possible new UIDs:
  1726. * - a real UID when storing a message to a UIDPLUS mailbox
  1727. * - -2 when storing a message to a dumb mailbox
  1728. * - -1 when not actually storing a message */
  1729. if (srec->uid[t] != uid) {
  1730. debug( " -> new UID %d on %s\n", uid, str_ms[t] );
  1731. jFprintf( svars, "%c %d %d %d\n", "<>"[t], srec->uid[M], srec->uid[S], uid );
  1732. srec->uid[t] = uid;
  1733. srec->tuid[0] = 0;
  1734. }
  1735. if (t == S && svars->mmaxxuid < srec->uid[M]) {
  1736. /* If we have so many new messages that some of them are instantly expired,
  1737. * but some are still propagated because they are important, we need to
  1738. * ensure explicitly that the bulk fetch limit is upped. */
  1739. svars->mmaxxuid = INT_MAX;
  1740. if (svars->smaxxuid < srec->uid[S] - 1) {
  1741. svars->smaxxuid = srec->uid[S] - 1;
  1742. jFprintf( svars, "! %d\n", svars->smaxxuid );
  1743. }
  1744. }
  1745. }
  1746. static void msgs_found_new( int sts, void *aux );
  1747. static void msgs_new_done( sync_vars_t *svars, int t );
  1748. static void sync_close( sync_vars_t *svars, int t );
  1749. static void
  1750. msgs_copied( sync_vars_t *svars, int t )
  1751. {
  1752. message_t *tmsg;
  1753. sync_rec_t *srec;
  1754. copy_vars_t *cv;
  1755. if (svars->state[t] & ST_SENDING_NEW)
  1756. return;
  1757. sync_ref( svars );
  1758. if (!(svars->state[t] & ST_SENT_NEW)) {
  1759. for (tmsg = svars->new_msgs[t]; tmsg; tmsg = tmsg->next) {
  1760. if ((srec = tmsg->srec) && srec->tuid[0]) {
  1761. if (svars->drv[t]->memory_usage( svars->ctx[t] ) >= BufferLimit) {
  1762. svars->new_msgs[t] = tmsg;
  1763. goto out;
  1764. }
  1765. new_total[t]++;
  1766. stats();
  1767. svars->new_pending[t]++;
  1768. svars->state[t] |= ST_SENDING_NEW;
  1769. cv = nfmalloc( sizeof(*cv) );
  1770. cv->cb = msg_copied;
  1771. cv->aux = AUX;
  1772. cv->srec = srec;
  1773. cv->msg = tmsg;
  1774. copy_msg( cv );
  1775. svars->state[t] &= ~ST_SENDING_NEW;
  1776. if (check_cancel( svars ))
  1777. goto out;
  1778. }
  1779. }
  1780. svars->state[t] |= ST_SENT_NEW;
  1781. }
  1782. if (svars->new_pending[t])
  1783. goto out;
  1784. if (svars->maxuid[1-t] != svars->newmaxuid[1-t]) {
  1785. svars->maxuid[1-t] = svars->newmaxuid[1-t];
  1786. jFprintf( svars, "S %d\n", 1-t );
  1787. }
  1788. sync_close( svars, 1-t );
  1789. if (check_cancel( svars ))
  1790. goto out;
  1791. if (svars->state[t] & ST_FIND_NEW) {
  1792. debug( "finding just copied messages on %s\n", str_ms[t] );
  1793. svars->drv[t]->find_new_msgs( svars->ctx[t], svars->newuid[t], msgs_found_new, AUX );
  1794. } else {
  1795. msgs_new_done( svars, t );
  1796. }
  1797. out:
  1798. sync_deref( svars );
  1799. }
  1800. static void
  1801. msgs_found_new( int sts, void *aux )
  1802. {
  1803. SVARS_CHECK_RET;
  1804. switch (sts) {
  1805. case DRV_OK:
  1806. debug( "matching just copied messages on %s\n", str_ms[t] );
  1807. break;
  1808. default:
  1809. warn( "Warning: cannot find newly stored messages on %s.\n", str_ms[t] );
  1810. break;
  1811. }
  1812. match_tuids( svars, t );
  1813. msgs_new_done( svars, t );
  1814. }
  1815. static void
  1816. msgs_new_done( sync_vars_t *svars, int t )
  1817. {
  1818. svars->state[t] |= ST_FOUND_NEW;
  1819. sync_close( svars, t );
  1820. }
  1821. static void
  1822. flags_set( int sts, void *aux )
  1823. {
  1824. SVARS_CHECK_RET_VARS(flag_vars_t);
  1825. switch (sts) {
  1826. case DRV_OK:
  1827. if (vars->aflags & F_DELETED)
  1828. vars->srec->status |= S_DEL(t);
  1829. else if (vars->dflags & F_DELETED)
  1830. vars->srec->status &= ~S_DEL(t);
  1831. flags_set_p2( svars, vars->srec, t );
  1832. break;
  1833. }
  1834. free( vars );
  1835. flags_done[t]++;
  1836. stats();
  1837. svars->flags_pending[t]--;
  1838. msgs_flags_set( svars, t );
  1839. }
  1840. static void
  1841. flags_set_p2( sync_vars_t *svars, sync_rec_t *srec, int t )
  1842. {
  1843. if (srec->status & S_DELETE) {
  1844. debug( " pair(%d,%d): resetting %s UID\n", srec->uid[M], srec->uid[S], str_ms[1-t] );
  1845. jFprintf( svars, "%c %d %d 0\n", "><"[t], srec->uid[M], srec->uid[S] );
  1846. srec->uid[1-t] = 0;
  1847. } else {
  1848. int nflags = (srec->flags | srec->aflags[t]) & ~srec->dflags[t];
  1849. if (srec->flags != nflags) {
  1850. debug( " pair(%d,%d): updating flags (%u -> %u; %sed)\n", srec->uid[M], srec->uid[S], srec->flags, nflags, str_hl[t] );
  1851. srec->flags = nflags;
  1852. jFprintf( svars, "* %d %d %u\n", srec->uid[M], srec->uid[S], nflags );
  1853. }
  1854. if (t == S) {
  1855. uint nex = (srec->status / S_NEXPIRE) & 1;
  1856. if (nex != ((srec->status / S_EXPIRED) & 1)) {
  1857. if (nex && (svars->smaxxuid < srec->uid[S]))
  1858. svars->smaxxuid = srec->uid[S];
  1859. jFprintf( svars, "/ %d %d\n", srec->uid[M], srec->uid[S] );
  1860. debug( " pair(%d,%d): expired %d (commit)\n", srec->uid[M], srec->uid[S], nex );
  1861. srec->status = (srec->status & ~S_EXPIRED) | (nex * S_EXPIRED);
  1862. } else if (nex != ((srec->status / S_EXPIRE) & 1)) {
  1863. jFprintf( svars, "\\ %d %d\n", srec->uid[M], srec->uid[S] );
  1864. debug( " pair(%d,%d): expire %d (cancel)\n", srec->uid[M], srec->uid[S], nex );
  1865. srec->status = (srec->status & ~S_EXPIRE) | (nex * S_EXPIRE);
  1866. }
  1867. }
  1868. }
  1869. }
  1870. typedef struct {
  1871. void *aux;
  1872. message_t *msg;
  1873. } trash_vars_t;
  1874. static void msg_trashed( int sts, void *aux );
  1875. static void msg_rtrashed( int sts, int uid, copy_vars_t *vars );
  1876. static void
  1877. msgs_flags_set( sync_vars_t *svars, int t )
  1878. {
  1879. message_t *tmsg;
  1880. trash_vars_t *tv;
  1881. copy_vars_t *cv;
  1882. if (!(svars->state[t] & ST_SENT_FLAGS) || svars->flags_pending[t])
  1883. return;
  1884. sync_ref( svars );
  1885. if ((svars->chan->ops[t] & OP_EXPUNGE) &&
  1886. (svars->ctx[t]->conf->trash || (svars->ctx[1-t]->conf->trash && svars->ctx[1-t]->conf->trash_remote_new))) {
  1887. debug( "trashing in %s\n", str_ms[t] );
  1888. for (tmsg = svars->ctx[t]->msgs; tmsg; tmsg = tmsg->next)
  1889. if ((tmsg->flags & F_DELETED) && !find_int_array( svars->trashed_msgs[t].array, tmsg->uid ) &&
  1890. (t == M || !tmsg->srec || !(tmsg->srec->status & (S_EXPIRE|S_EXPIRED)))) {
  1891. if (svars->ctx[t]->conf->trash) {
  1892. if (!svars->ctx[t]->conf->trash_only_new || !tmsg->srec || tmsg->srec->uid[1-t] < 0) {
  1893. debug( "%s: trashing message %d\n", str_ms[t], tmsg->uid );
  1894. trash_total[t]++;
  1895. stats();
  1896. svars->trash_pending[t]++;
  1897. tv = nfmalloc( sizeof(*tv) );
  1898. tv->aux = AUX;
  1899. tv->msg = tmsg;
  1900. svars->drv[t]->trash_msg( svars->ctx[t], tmsg, msg_trashed, tv );
  1901. if (check_cancel( svars ))
  1902. goto out;
  1903. } else
  1904. debug( "%s: not trashing message %d - not new\n", str_ms[t], tmsg->uid );
  1905. } else {
  1906. if (!tmsg->srec || tmsg->srec->uid[1-t] < 0) {
  1907. if (tmsg->size <= svars->ctx[1-t]->conf->max_size) {
  1908. debug( "%s: remote trashing message %d\n", str_ms[t], tmsg->uid );
  1909. trash_total[t]++;
  1910. stats();
  1911. svars->trash_pending[t]++;
  1912. cv = nfmalloc( sizeof(*cv) );
  1913. cv->cb = msg_rtrashed;
  1914. cv->aux = INV_AUX;
  1915. cv->srec = 0;
  1916. cv->msg = tmsg;
  1917. copy_msg( cv );
  1918. if (check_cancel( svars ))
  1919. goto out;
  1920. } else
  1921. debug( "%s: not remote trashing message %d - too big\n", str_ms[t], tmsg->uid );
  1922. } else
  1923. debug( "%s: not remote trashing message %d - not new\n", str_ms[t], tmsg->uid );
  1924. }
  1925. }
  1926. }
  1927. svars->state[t] |= ST_SENT_TRASH;
  1928. sync_close( svars, t );
  1929. out:
  1930. sync_deref( svars );
  1931. }
  1932. static void
  1933. msg_trashed( int sts, void *aux )
  1934. {
  1935. trash_vars_t *vars = (trash_vars_t *)aux;
  1936. DECL_SVARS;
  1937. if (sts == DRV_MSG_BAD)
  1938. sts = DRV_BOX_BAD;
  1939. if (check_ret( sts, vars->aux ))
  1940. return;
  1941. INIT_SVARS(vars->aux);
  1942. debug( " -> trashed %s %d\n", str_ms[t], vars->msg->uid );
  1943. jFprintf( svars, "T %d %d\n", t, vars->msg->uid );
  1944. free( vars );
  1945. trash_done[t]++;
  1946. stats();
  1947. svars->trash_pending[t]--;
  1948. sync_close( svars, t );
  1949. }
  1950. static void
  1951. msg_rtrashed( int sts, int uid ATTR_UNUSED, copy_vars_t *vars )
  1952. {
  1953. SVARS_CHECK_CANCEL_RET;
  1954. switch (sts) {
  1955. case SYNC_OK:
  1956. case SYNC_NOGOOD: /* the message is gone or heavily busted */
  1957. break;
  1958. default:
  1959. cancel_sync( svars );
  1960. free( vars );
  1961. return;
  1962. }
  1963. t ^= 1;
  1964. debug( " -> remote trashed %s %d\n", str_ms[t], vars->msg->uid );
  1965. jFprintf( svars, "T %d %d\n", t, vars->msg->uid );
  1966. free( vars );
  1967. trash_done[t]++;
  1968. stats();
  1969. svars->trash_pending[t]--;
  1970. sync_close( svars, t );
  1971. }
  1972. static void box_closed( int sts, void *aux );
  1973. static void box_closed_p2( sync_vars_t *svars, int t );
  1974. static void
  1975. sync_close( sync_vars_t *svars, int t )
  1976. {
  1977. if ((~svars->state[t] & (ST_FOUND_NEW|ST_SENT_TRASH)) || svars->trash_pending[t] ||
  1978. !(svars->state[1-t] & ST_SENT_NEW) || svars->new_pending[1-t])
  1979. return;
  1980. if (svars->state[t] & ST_CLOSING)
  1981. return;
  1982. svars->state[t] |= ST_CLOSING;
  1983. if ((svars->chan->ops[t] & OP_EXPUNGE) /*&& !(svars->state[t] & ST_TRASH_BAD)*/) {
  1984. debug( "expunging %s\n", str_ms[t] );
  1985. svars->drv[t]->close_box( svars->ctx[t], box_closed, AUX );
  1986. } else {
  1987. box_closed_p2( svars, t );
  1988. }
  1989. }
  1990. static void
  1991. box_closed( int sts, void *aux )
  1992. {
  1993. SVARS_CHECK_RET;
  1994. svars->state[t] |= ST_DID_EXPUNGE;
  1995. box_closed_p2( svars, t );
  1996. }
  1997. static void
  1998. box_closed_p2( sync_vars_t *svars, int t )
  1999. {
  2000. sync_rec_t *srec;
  2001. int minwuid;
  2002. svars->state[t] |= ST_CLOSED;
  2003. if (!(svars->state[1-t] & ST_CLOSED))
  2004. return;
  2005. if (((svars->state[M] | svars->state[S]) & ST_DID_EXPUNGE) || svars->chan->max_messages) {
  2006. debug( "purging obsolete entries\n" );
  2007. minwuid = INT_MAX;
  2008. if (svars->chan->max_messages) {
  2009. debug( " max expired slave uid is %d\n", svars->smaxxuid );
  2010. for (srec = svars->srecs; srec; srec = srec->next) {
  2011. if (srec->status & S_DEAD)
  2012. continue;
  2013. if (!((srec->uid[S] <= 0 || ((srec->status & S_DEL(S)) && (svars->state[S] & ST_DID_EXPUNGE))) &&
  2014. (srec->uid[M] <= 0 || ((srec->status & S_DEL(M)) && (svars->state[M] & ST_DID_EXPUNGE)) || (srec->status & S_EXPIRED))) &&
  2015. svars->smaxxuid < srec->uid[S] && minwuid > srec->uid[M])
  2016. minwuid = srec->uid[M];
  2017. }
  2018. debug( " min non-orphaned master uid is %d\n", minwuid );
  2019. }
  2020. for (srec = svars->srecs; srec; srec = srec->next) {
  2021. if (srec->status & S_DEAD)
  2022. continue;
  2023. if (srec->uid[S] <= 0 || ((srec->status & S_DEL(S)) && (svars->state[S] & ST_DID_EXPUNGE))) {
  2024. if (srec->uid[M] <= 0 || ((srec->status & S_DEL(M)) && (svars->state[M] & ST_DID_EXPUNGE)) ||
  2025. ((srec->status & S_EXPIRED) && svars->maxuid[M] >= srec->uid[M] && minwuid > srec->uid[M])) {
  2026. debug( " -> killing (%d,%d)\n", srec->uid[M], srec->uid[S] );
  2027. srec->status = S_DEAD;
  2028. jFprintf( svars, "- %d %d\n", srec->uid[M], srec->uid[S] );
  2029. } else if (srec->uid[S] > 0) {
  2030. debug( " -> orphaning (%d,[%d])\n", srec->uid[M], srec->uid[S] );
  2031. jFprintf( svars, "> %d %d 0\n", srec->uid[M], srec->uid[S] );
  2032. srec->uid[S] = 0;
  2033. }
  2034. } else if (srec->uid[M] > 0 && ((srec->status & S_DEL(M)) && (svars->state[M] & ST_DID_EXPUNGE))) {
  2035. debug( " -> orphaning ([%d],%d)\n", srec->uid[M], srec->uid[S] );
  2036. jFprintf( svars, "< %d %d 0\n", srec->uid[M], srec->uid[S] );
  2037. srec->uid[M] = 0;
  2038. }
  2039. }
  2040. }
  2041. save_state( svars );
  2042. sync_bail( svars );
  2043. }
  2044. static void
  2045. sync_bail( sync_vars_t *svars )
  2046. {
  2047. sync_rec_t *srec, *nsrec;
  2048. free( svars->trashed_msgs[M].array.data );
  2049. free( svars->trashed_msgs[S].array.data );
  2050. for (srec = svars->srecs; srec; srec = nsrec) {
  2051. nsrec = srec->next;
  2052. free( srec );
  2053. }
  2054. if (svars->lfd >= 0) {
  2055. unlink( svars->lname );
  2056. close( svars->lfd );
  2057. }
  2058. sync_bail2( svars );
  2059. }
  2060. static void
  2061. sync_bail2( sync_vars_t *svars )
  2062. {
  2063. free( svars->lname );
  2064. free( svars->nname );
  2065. free( svars->jname );
  2066. free( svars->dname );
  2067. sync_bail3( svars );
  2068. }
  2069. static void
  2070. sync_bail3( sync_vars_t *svars )
  2071. {
  2072. free( svars->box_name[M] );
  2073. free( svars->box_name[S] );
  2074. sync_deref( svars );
  2075. }
  2076. static void
  2077. sync_deref( sync_vars_t *svars )
  2078. {
  2079. if (!--svars->ref_count) {
  2080. void (*cb)( int sts, void *aux ) = svars->cb;
  2081. void *aux = svars->aux;
  2082. int ret = svars->ret;
  2083. free( svars );
  2084. cb( ret, aux );
  2085. }
  2086. }