Make IMAP recent claims atomic

include/nmlib.h

@@ -164,6 +164,8 @@ int NMAPSetDecimalProperty(Connection *conn, uint64_t guid, const char *property
 int NMAPReadHexadecimalPropertyResponse(Connection *conn, const char *propertyName, unsigned long *propertyValue);
 int NMAPGetHexadecimalProperty(Connection *conn, uint64_t guid, const char *propertyName, unsigned long *propertyValue);
 int NMAPSetHexadecimalProperty(Connection *conn, uint64_t guid, const char *propertyName, unsigned long propertyValue);
+int NMAPClaimImapRecent(Connection *conn, uint64_t guid, uint32_t requestedEnd,
+                       uint32_t *previousEnd, uint32_t *storedEnd);
 
 int NMAPCreateCollection(Connection *conn,
                          const char *collectionName,

src/agents/imap/event.c

@@ -547,45 +547,44 @@ SendRememberedEvents(ImapSession *session, unsigned long typesAllowed)
                 
     if ((events->remembered & STORE_EVENT_NEW) && (typesAllowed & STORE_EVENT_NEW)) {
         uint32_t recentUid;
-        BOOL recentLocked = !selectedFolder->readOnly;
+        uint32_t requestedEnd = 0;
+        unsigned long eventIndex;
 
         sequenceChanged = TRUE;
         /*
-         * Refresh the persistent threshold before loading new messages.  The
-         * first read-write session to process the event keeps its local
-         * \Recent flags and advances the shared threshold; later sessions
-         * therefore cannot report the same messages as recent.
+         * Claim only the UIDs present in this reconciled event batch.  A
+         * delivery committed after that snapshot remains available for the
+         * next event batch instead of being silently claimed here.
          */
-        if (recentLocked) FolderRecentLock();
-        ccode = FolderGetRecentUid(storeConn, selectedFolder->info,
-                                   &recentUid);
-        if (ccode != STATUS_CONTINUE) {
-            if (recentLocked) FolderRecentUnlock();
-            return ccode;
+        for (eventIndex = 0; eventIndex < events->newCount; eventIndex++) {
+            if (events->new[eventIndex].uid >= requestedEnd)
+                requestedEnd = events->new[eventIndex].uid + 1;
+        }
+        if (selectedFolder->readOnly) {
+            ccode = FolderGetRecentUid(storeConn, selectedFolder->info,
+                                       &recentUid);
+            if (ccode != STATUS_CONTINUE)
+                return ccode;
+            selectedFolder->info->uidRecent = recentUid;
+            selectedFolder->info->uidRecentEnd = requestedEnd;
+        } else {
+            ccode = FolderClaimRecentUids(storeConn, selectedFolder->info,
+                                          requestedEnd);
+            if (ccode != STATUS_CONTINUE)
+                return ccode;
         }
-        selectedFolder->info->uidRecent = recentUid;
         ccode = MessageListAddNewMessages(storeConn, selectedFolder,
                                           events->new, events->newCount);
-        if (ccode != STATUS_CONTINUE) {
-            if (recentLocked) FolderRecentUnlock();
+        if (ccode != STATUS_CONTINUE)
             return ccode;
-        }
         /*
          * A newly delivered message can already carry keywords.  Announce
          * the expanded mailbox FLAGS set before any later FETCH can expose
          * such a keyword to this selected client.
          */
         ccode = SendMailboxFlags(clientConn, selectedFolder);
-        if (ccode != STATUS_CONTINUE) {
-            if (recentLocked) FolderRecentUnlock();
+        if (ccode != STATUS_CONTINUE)
             return ccode;
-        }
-        if (!selectedFolder->readOnly) {
-            ccode = FolderSetRecentUid(storeConn, selectedFolder->info);
-            FolderRecentUnlock();
-            recentLocked = FALSE;
-            if (ccode != STATUS_CONTINUE) return ccode;
-        }
         events->newCount = 0;
         events->remembered &= ~STORE_EVENT_NEW;
     }

src/agents/imap/imapd.c

@@ -321,30 +321,30 @@ FolderListShutdown(ImapSession *session)
 }
 
 long
-FolderSetRecentUid(Connection *storeConn, FolderInformation *folder)
+FolderClaimRecentUids(Connection *storeConn, FolderInformation *folder,
+                      uint32_t requestedEnd)
 {
     long ccode;
+    uint32_t previousEnd;
+    uint32_t storedEnd;
 
-    /* Set the new recent uid equal to the uidNext */
-    ccode = NMAPSetHexadecimalProperty(storeConn, folder->guid, "imap.recentuid", folder->uidNext);
-    if (ccode == 1000)  {
+    ccode = NMAPClaimImapRecent(storeConn, folder->guid, requestedEnd,
+                                &previousEnd, &storedEnd);
+    if (ccode == 1000) {
+        if (storedEnd < previousEnd || storedEnd < requestedEnd)
+            return STATUS_NMAP_PROTOCOL_ERROR;
+        folder->uidRecent = previousEnd;
+        /*
+         * If another session already advanced beyond requestedEnd, this
+         * session owns no part of that interval.
+         */
+        folder->uidRecentEnd =
+            previousEnd < requestedEnd ? requestedEnd : previousEnd;
         return(STATUS_CONTINUE);
     }
     return(CheckForNMAPCommError(ccode));
 }
 
-void
-FolderRecentLock(void)
-{
-    XplMutexLock(Imap.recentLock);
-}
-
-void
-FolderRecentUnlock(void)
-{
-    XplMutexUnlock(Imap.recentLock);
-}
-
 __inline static long
 FolderWatchAdd(Connection *storeConn, FolderInformation *folder, const char *eventString)
 {
@@ -515,9 +515,17 @@ FolderOpen(Connection *storeConn, OpenedFolder *openFolder, FolderInformation *f
     long ccode;
     uint32_t highestUid = 0;
 
-    ccode = FolderGetRecentUid(storeConn, folder, &folder->uidRecent);
+    ccode = FolderGetHighestUid(storeConn, folder, &highestUid);
     if (ccode == STATUS_CONTINUE) {
-        ccode = FolderGetHighestUid(storeConn, folder, &highestUid);
+        folder->uidNext = highestUid + 1;
+        if (readOnly) {
+            ccode = FolderGetRecentUid(storeConn, folder,
+                                       &folder->uidRecent);
+            folder->uidRecentEnd = folder->uidNext;
+        } else {
+            ccode = FolderClaimRecentUids(storeConn, folder,
+                                          folder->uidNext);
+        }
         if (ccode == STATUS_CONTINUE) {
             openFolder->info = folder;
             ccode = MessageListLoad(storeConn, openFolder);
@@ -557,31 +565,13 @@ FolderSelect(ImapSession *session, char *folderName, BOOL readOnly)
 
     if ((ccode = FolderGetByName(session, folderName, &folder)) == STATUS_CONTINUE) {
         if ((ccode = FolderWatchAdd(session->store.conn, folder, "FLAGS MODIFIED NEW DELETED")) == STATUS_CONTINUE) {
-            if (!readOnly) FolderRecentLock();
             ccode = FolderOpen(session->store.conn,
                                &session->folder.selected, folder, readOnly);
             if (ccode == STATUS_CONTINUE) {
-                /*
-                 * A read-write SELECT claims the current \Recent set for this
-                 * session.  Keep the session-local flags loaded by FolderOpen,
-                 * but advance the persistent threshold immediately so STATUS
-                 * and other sessions cannot claim the same messages.
-                 */
-                if (!readOnly &&
-                    (ccode = FolderSetRecentUid(session->store.conn, folder)) !=
-                        STATUS_CONTINUE) {
-                    FolderRecentUnlock();
-                    FolderClose(&session->folder.selected);
-                    FolderWatchRemove(session->store.conn, folder);
-                    EventsFree(&session->folder.selected.events);
-                    return ccode;
-                }
-                if (!readOnly) FolderRecentUnlock();
                 session->client.state = STATE_SELECTED;
                 ImapSearchResultsReset(session);
                 return(STATUS_CONTINUE);
             }
-            if (!readOnly) FolderRecentUnlock();
             FolderWatchRemove(session->store.conn, folder);
             EventsFree(&(session->folder.selected.events));
         }
@@ -4486,7 +4476,6 @@ InitializeImapGlobals()
     /* Initialize the Busy List and Semaphore */
     Imap.list_Busy = NULL;
     XplOpenLocalSemaphore(Imap.sem_Busy, 1);
-    XplMutexInit(Imap.recentLock);
 
     /* Global allocations */
     BongoKeywordIndexCreateFromTable(Imap.command.index, ImapProtocolCommands, .name, TRUE);
@@ -4524,7 +4513,6 @@ FreeImapGlobals()
     CommandFetchCleanup();
     FreeReturnValueIndex(Imap.command.returnValueIndex);
     BongoKeywordIndexFree(Imap.command.index);
-    XplMutexDestroy(Imap.recentLock);
     return(TRUE);
 }
 

src/agents/imap/imapd.h

@@ -265,6 +265,7 @@ typedef struct {
     uint64_t guid;                                          /* folder id                        */
     uint32_t uidNext;                                       /* the next expected uid            */
     uint32_t uidRecent;                                     /* the first recent uid             */
+    uint32_t uidRecentEnd;                                  /* exclusive end of recent claim    */
     struct {
         char *utf7;                                         /* utf7 encoding of the folder name */
         char *utf8;                                         /* utf8 encoding of the folder name */
@@ -368,7 +369,8 @@ typedef struct {
 
 long FolderGetRecentUid(Connection *storeConn, FolderInformation *folder,
                         uint32_t *recentUid);
-long FolderSetRecentUid(Connection *storeConn, FolderInformation *folder);
+long FolderClaimRecentUids(Connection *storeConn, FolderInformation *folder,
+                           uint32_t requestedEnd);
 
 typedef struct {
     struct {
@@ -442,7 +444,6 @@ typedef struct {
 
     BongoList *list_Busy;      /* Singly linked list of sessions that we should update every 10 seconds */
     XplSemaphore sem_Busy;      /* Semaphore protecting the busy list */
-    XplMutex recentLock;        /* serializes the shared \Recent claim */
     
     void *logHandle;
     XplAtomic exiting;
@@ -464,8 +465,6 @@ long GetMessageRange(MessageInformation *message, unsigned long messageCount, ch
 
 long FolderOpen(Connection *storeConn, OpenedFolder *openFolder, FolderInformation *folder, BOOL readOnly);
 long FolderListLoad(ImapSession *session);
-void FolderRecentLock(void);
-void FolderRecentUnlock(void);
 long FolderListInitialize(ImapSession *session);
 long MessageListLoad(Connection *conn, OpenedFolder *selected);
 

src/agents/imap/inline.h

@@ -819,7 +819,8 @@ MessageListAddMessage(OpenedFolder *folder, char *response,
                     keywords && keywords[0] ? MemStrdup(keywords) : NULL;
                 if (keywords && keywords[0] && !currentMessage->keywords)
                     return STATUS_MEMORY_ERROR;
-                if (currentMessage->uid >= folder->info->uidRecent) {
+                if (currentMessage->uid >= folder->info->uidRecent &&
+                    currentMessage->uid < folder->info->uidRecentEnd) {
                     currentMessage->flags |= STORE_MSG_FLAG_RECENT;
                     folder->recentCount++;
                 }

src/agents/store/command.c

@@ -142,6 +142,8 @@ StoreSetupCommands()
         BongoHashtablePutNoReplace(CommandTable, "RENAME", (void *) STORE_COMMAND_RENAME) ||
         BongoHashtablePutNoReplace(CommandTable, "REMOVE", (void *) STORE_COMMAND_REMOVE) ||
         BongoHashtablePutNoReplace(CommandTable, "SEARCH", (void *) STORE_COMMAND_SEARCH) ||
+        BongoHashtablePutNoReplace(CommandTable, "IMAPRECENT",
+                                  (void *) STORE_COMMAND_IMAPRECENT) ||
         BongoHashtablePutNoReplace(CommandTable, "WATCH", (void *) STORE_COMMAND_WATCH) ||
         BongoHashtablePutNoReplace(CommandTable, "REPAIR", (void *) STORE_COMMAND_REPAIR) ||
 
@@ -846,6 +848,28 @@ StoreCommandLoop(StoreClient *client)
                                         props, int3);
             break;
 
+        case STORE_COMMAND_IMAPRECENT:
+            /*
+             * IMAPRECENT <collection> <exclusive-end-uid>
+             *
+             * Atomically claim a half-open UID interval for IMAP4rev1's
+             * session-specific \Recent flag.
+             */
+            {
+                uint32_t requestedEnd;
+
+                if (TOKEN_OK == (ccode = RequireStore(client)) &&
+                    TOKEN_OK == (ccode = CheckTokC(client, n, 3, 3)) &&
+                    TOKEN_OK == (ccode =
+                        ParseCollection(client, tokens[1], &object)) &&
+                    TOKEN_OK == (ccode =
+                        ParseHexU32(client, tokens[2], &requestedEnd))) {
+                    ccode = StoreCommandIMAPRECENT(client, &object,
+                                                   requestedEnd);
+                }
+            }
+            break;
+
         case STORE_COMMAND_LINK:
             /* LINK <document> <document> 
              * abuses collection - first argument should be another document*/
@@ -2805,6 +2829,120 @@ StoreCommandPROPSET(StoreClient *client,
 	return ccode;
 }
 
+CCode
+StoreCommandIMAPRECENT(StoreClient *client, StoreObject *collection,
+                       uint32_t requestedEnd)
+{
+    static const char propertyName[] = "imap.recentuid";
+    MsgSQLStatement selectStmt;
+    MsgSQLStatement deleteStmt;
+    MsgSQLStatement insertStmt;
+    BOOL selectPrepared = FALSE;
+    BOOL deletePrepared = FALSE;
+    BOOL insertPrepared = FALSE;
+    uint32_t previousEnd = 0;
+    uint32_t storedEnd;
+    char value[16];
+    char *end = NULL;
+    unsigned long parsedEnd;
+    int result;
+    int auth;
+
+    CHECK_NOT_READONLY(client)
+
+    auth = StoreObjectCheckAuthorization(
+        client, collection, STORE_PRIV_READ_PROPS | STORE_PRIV_WRITE_PROPS);
+    if (auth)
+        return ConnWriteStr(client->conn, MSG4240NOPERMISSION);
+
+    if (!LogicalLockGain(client, collection, LLOCK_READWRITE,
+                         "StoreCommandIMAPRECENT"))
+        return ConnWriteStr(client->conn, MSG4120BOXLOCKED);
+
+    memset(&selectStmt, 0, sizeof(selectStmt));
+    memset(&deleteStmt, 0, sizeof(deleteStmt));
+    memset(&insertStmt, 0, sizeof(insertStmt));
+
+    if (MsgSQLBeginTransaction(client->storedb))
+        goto abort;
+
+    if (!MsgSQLPrepare(
+            client->storedb,
+            "SELECT value FROM properties "
+            "WHERE guid = ?1 AND name = ?2 LIMIT 1;",
+            &selectStmt))
+        goto abort;
+    selectPrepared = TRUE;
+    MsgSQLBindInt64(&selectStmt, 1, collection->guid);
+    MsgSQLBindString(&selectStmt, 2, propertyName, FALSE);
+    result = MsgSQLResults(client->storedb, &selectStmt);
+    if (result < 0)
+        goto abort;
+    if (result > 0) {
+        if (MsgSQLResultText(&selectStmt, 0, value, sizeof(value)) < 0)
+            goto abort;
+        errno = 0;
+        parsedEnd = strtoul(value, &end, 16);
+        if (errno == ERANGE || !end || *end || parsedEnd > UINT32_MAX)
+            goto abort;
+        previousEnd = (uint32_t)parsedEnd;
+    }
+    MsgSQLFinalize(&selectStmt);
+    selectPrepared = FALSE;
+
+    storedEnd = requestedEnd > previousEnd ? requestedEnd : previousEnd;
+    if (storedEnd != previousEnd) {
+        if (!MsgSQLPrepare(
+                client->storedb,
+                "DELETE FROM properties WHERE guid = ?1 AND name = ?2;",
+                &deleteStmt))
+            goto abort;
+        deletePrepared = TRUE;
+        MsgSQLBindInt64(&deleteStmt, 1, collection->guid);
+        MsgSQLBindString(&deleteStmt, 2, propertyName, FALSE);
+        if (MsgSQLExecute(client->storedb, &deleteStmt) != 0)
+            goto abort;
+        MsgSQLFinalize(&deleteStmt);
+        deletePrepared = FALSE;
+
+        snprintf(value, sizeof(value), "%x", storedEnd);
+        if (!MsgSQLPrepare(
+                client->storedb,
+                "INSERT INTO properties (guid, intprop, name, value) "
+                "VALUES (?1, NULL, ?2, ?3);",
+                &insertStmt))
+            goto abort;
+        insertPrepared = TRUE;
+        MsgSQLBindInt64(&insertStmt, 1, collection->guid);
+        MsgSQLBindString(&insertStmt, 2, propertyName, FALSE);
+        MsgSQLBindString(&insertStmt, 3, value, FALSE);
+        if (MsgSQLExecute(client->storedb, &insertStmt) != 0)
+            goto abort;
+        MsgSQLFinalize(&insertStmt);
+        insertPrepared = FALSE;
+    }
+
+    if (MsgSQLCommitTransaction(client->storedb))
+        goto abort;
+    LogicalLockRelease(client, collection, LLOCK_READWRITE,
+                       "StoreCommandIMAPRECENT");
+
+    return ConnWriteF(client->conn, "1000 OK %x %x\r\n",
+                      previousEnd, storedEnd);
+
+abort:
+    if (selectPrepared)
+        MsgSQLFinalize(&selectStmt);
+    if (deletePrepared)
+        MsgSQLFinalize(&deleteStmt);
+    if (insertPrepared)
+        MsgSQLFinalize(&insertStmt);
+    MsgSQLAbortTransaction(client->storedb);
+    LogicalLockRelease(client, collection, LLOCK_READWRITE,
+                       "StoreCommandIMAPRECENT");
+    return ConnWriteStr(client->conn, MSG5005DBLIBERR);
+}
+
 
 /* returns: -1, causing storecommandloop to exit */
 CCode

src/agents/store/command.h

@@ -54,6 +54,7 @@ typedef enum {
     STORE_COMMAND_REPAIR,
     STORE_COMMAND_SEARCH,
     STORE_COMMAND_STATUS,
+    STORE_COMMAND_IMAPRECENT,
     STORE_COMMAND_WATCH,
 
     /* document commands */
@@ -232,6 +233,9 @@ CCode StoreCommandPROPGET(StoreClient *client, StoreObject *object,
 CCode StoreCommandPROPSET(StoreClient *client, StoreObject *object, 
                         StorePropInfo *prop, int size);
 
+CCode StoreCommandIMAPRECENT(StoreClient *client, StoreObject *collection,
+                             uint32_t requestedEnd);
+
 CCode StoreCommandQUIT(StoreClient *client);
 
 CCode StoreCommandREAD(StoreClient *client, StoreObject *object, 

src/libs/connio/nmap.c

@@ -1020,6 +1020,32 @@ NMAPSetHexadecimalProperty(Connection *conn, uint64_t guid, const char *property
     return(-1);
 }
 
+int
+NMAPClaimImapRecent(Connection *conn, uint64_t guid, uint32_t requestedEnd,
+                    uint32_t *previousEnd, uint32_t *storedEnd)
+{
+    char response[64];
+    unsigned int previous;
+    unsigned int stored;
+    char trailing;
+    int ccode;
+
+    if (!previousEnd || !storedEnd)
+        return -1;
+
+    ccode = NMAPRunCommandF(conn, response, sizeof(response),
+                            "IMAPRECENT " STORE_GUID_FMT " %x\r\n",
+                            guid, requestedEnd);
+    if (ccode != 1000)
+        return ccode;
+    if (sscanf(response, "OK %x %x %c", &previous, &stored, &trailing) != 2)
+        return -1;
+
+    *previousEnd = (uint32_t)previous;
+    *storedEnd = (uint32_t)stored;
+    return 1000;
+}
+
 BongoCalObject *
 NMAPGetEvents(Connection *conn, const char *calendar, BongoCalTime start, BongoCalTime end)
 {