fix(GODT-1621): Fix race condition with temporary IDs

LBeernaertProton · LBeernaertProton · commit 77ef5f84a4fc · 2022-06-28T16:31:01.000+02:00
Currently it is possible to have temporary IDs for mailboxes and
messages to become stale. It is possible, due to concurrency, that a
session pushes an operation which still uses a temporary ID that is no
longer valid.

Rather than locking the entire queue and updating all the remaining
entries, we now maintain two separate translation tables which are
evicted via separate operations.

This reduces the amount of time the queue spends in a locking state and
also guarantees that by the time the removal of the temporary ID is
processed by the operation queue no other operations exist that still
refer to this temporary ID.
diff --git a/go.mod b/go.mod
@@ -12,7 +12,6 @@ require (
 	github.com/golang/mock v1.6.0
 	github.com/google/uuid v1.3.0
 	github.com/mattn/go-sqlite3 v1.14.10
-	github.com/pkg/errors v0.9.1
 	github.com/sirupsen/logrus v1.8.1
 	github.com/stretchr/testify v1.7.1-0.20210427113832-6241f9ab9942
 	golang.org/x/exp v0.0.0-20220518171630-0b5c67f07fdf
@@ -33,6 +32,7 @@ require (
 	github.com/hashicorp/hcl/v2 v2.10.0 // indirect
 	github.com/kr/pretty v0.3.0 // indirect
 	github.com/mitchellh/go-wordwrap v0.0.0-20150314170334-ad45545899c7 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rogpeppe/go-internal v1.8.0 // indirect
 	github.com/zclconf/go-cty v1.8.0 // indirect
diff --git a/internal/backend/user_updates.go b/internal/backend/user_updates.go
@@ -110,6 +110,10 @@ func (user *user) applyMailboxIDChanged(ctx context.Context, tx *ent.Tx, update
 		return err
 	}
 
+	if err := user.remote.FinishMailboxIDUpdate(update.OldID); err != nil {
+		logrus.WithError(err).Errorf("Call to FinishMailboxIDUpdate() failed")
+	}
+
 	return txUpdateMailboxID(ctx, tx, update.OldID, update.NewID)
 }
 
@@ -208,6 +212,10 @@ func (user *user) applyMessageIDChanged(ctx context.Context, tx *ent.Tx, update
 		return err
 	}
 
+	if err := user.remote.FinishMessageIDUpdate(update.OldID); err != nil {
+		logrus.WithError(err).Error("Call to FinishMessageIDUpdate() failed")
+	}
+
 	if err := user.store.Update(update.OldID, update.NewID); err != nil {
 		return err
 	}
diff --git a/internal/remote/operation.go b/internal/remote/operation.go
@@ -21,14 +21,6 @@ func (c *OperationBase) getConnMetadataID() ConnMetadataID {
 	return c.MetadataID
 }
 
-type mailboxOperation interface {
-	setMailboxID(tempID, mailboxID string)
-}
-
-type messageOperation interface {
-	setMessageID(tempID, messageID string)
-}
-
 // saveOps serializes the operation queue to a binary format to be serialized to disk.
 func saveOps(ops []operation) ([]byte, error) {
 	buf := new(bytes.Buffer)
diff --git a/internal/remote/operation_mailbox_delete.go b/internal/remote/operation_mailbox_delete.go
@@ -15,10 +15,4 @@ func (op *OpMailboxDelete) merge(other operation) (operation, bool) {
 	return nil, false
 }
 
-func (op *OpMailboxDelete) setMailboxID(tempID, mailboxID string) {
-	if op.MBoxID == tempID {
-		op.MBoxID = mailboxID
-	}
-}
-
 func (OpMailboxDelete) _isOperation() {}
diff --git a/internal/remote/operation_mailbox_update.go b/internal/remote/operation_mailbox_update.go
@@ -16,10 +16,4 @@ func (op *OpMailboxUpdate) merge(other operation) (operation, bool) {
 	return nil, false
 }
 
-func (op *OpMailboxUpdate) setMailboxID(tempID, mailboxID string) {
-	if op.MBoxID == tempID {
-		op.MBoxID = mailboxID
-	}
-}
-
 func (OpMailboxUpdate) _isOperation() {}
diff --git a/internal/remote/operation_message_add.go b/internal/remote/operation_message_add.go
@@ -28,18 +28,4 @@ func (op *OpMessageAdd) merge(other operation) (operation, bool) {
 	}
 }
 
-func (op *OpMessageAdd) setMailboxID(tempID, mboxID string) {
-	if op.MBoxID == tempID {
-		op.MBoxID = mboxID
-	}
-}
-
-func (op *OpMessageAdd) setMessageID(tempID, messageID string) {
-	for idx := range op.MessageIDs {
-		if op.MessageIDs[idx] == tempID {
-			op.MessageIDs[idx] = messageID
-		}
-	}
-}
-
 func (OpMessageAdd) _isOperation() {}
diff --git a/internal/remote/operation_message_create.go b/internal/remote/operation_message_create.go
@@ -24,10 +24,4 @@ func (op *OpMessageCreate) merge(other operation) (operation, bool) {
 	return nil, false
 }
 
-func (op *OpMessageCreate) setMailboxID(tempID, mailboxID string) {
-	if op.MBoxID == tempID {
-		op.MBoxID = mailboxID
-	}
-}
-
 func (OpMessageCreate) _isOperation() {}
diff --git a/internal/remote/operation_message_flagged.go b/internal/remote/operation_message_flagged.go
@@ -28,12 +28,4 @@ func (op *OpMessageFlagged) merge(other operation) (operation, bool) {
 	}
 }
 
-func (op *OpMessageFlagged) setMessageID(tempID, messageID string) {
-	for idx := range op.MessageIDs {
-		if op.MessageIDs[idx] == tempID {
-			op.MessageIDs[idx] = messageID
-		}
-	}
-}
-
 func (OpMessageFlagged) _isOperation() {}
diff --git a/internal/remote/operation_message_remove.go b/internal/remote/operation_message_remove.go
@@ -28,20 +28,6 @@ func (op *OpMessageRemove) merge(other operation) (operation, bool) {
 	}
 }
 
-func (op *OpMessageRemove) setMailboxID(tempID, mboxID string) {
-	if op.MBoxID == tempID {
-		op.MBoxID = mboxID
-	}
-}
-
-func (op *OpMessageRemove) setMessageID(tempID, messageID string) {
-	for idx := range op.MessageIDs {
-		if op.MessageIDs[idx] == tempID {
-			op.MessageIDs[idx] = messageID
-		}
-	}
-}
-
 func (op *OpMessageRemove) getConnMetadataID() ConnMetadataID {
 	return op.MetadataID
 }
diff --git a/internal/remote/operation_message_seen.go b/internal/remote/operation_message_seen.go
@@ -28,12 +28,4 @@ func (op *OpMessageSeen) merge(other operation) (operation, bool) {
 	}
 }
 
-func (op *OpMessageSeen) setMessageID(tempID, messageID string) {
-	for idx := range op.MessageIDs {
-		if op.MessageIDs[idx] == tempID {
-			op.MessageIDs[idx] = messageID
-		}
-	}
-}
-
 func (OpMessageSeen) _isOperation() {}
diff --git a/internal/remote/operation_temp_id.go b/internal/remote/operation_temp_id.go
@@ -0,0 +1,48 @@
+package remote
+
+import "encoding/gob"
+
+type OpRemMailboxTempID struct {
+	OperationBase
+	tempID string
+}
+
+type OpRemMessageTempID struct {
+	OperationBase
+	tempID string
+}
+
+func (op *OpRemMailboxTempID) merge(other operation) (operation, bool) {
+	switch other := other.(type) {
+	case *OpRemMailboxTempID:
+		if op.tempID != other.tempID || op.MetadataID != other.MetadataID {
+			return nil, false
+		}
+
+		return op, true
+	default:
+		return nil, false
+	}
+}
+
+func (OpRemMailboxTempID) _isOperation() {}
+
+func (op *OpRemMessageTempID) merge(other operation) (operation, bool) {
+	switch other := other.(type) {
+	case *OpRemMessageTempID:
+		if op.tempID != other.tempID || op.MetadataID != other.MetadataID {
+			return nil, false
+		}
+
+		return op, true
+	default:
+		return nil, false
+	}
+}
+
+func (OpRemMessageTempID) _isOperation() {}
+
+func init() {
+	gob.Register(&OpRemMailboxTempID{})
+	gob.Register(&OpRemMessageTempID{})
+}
diff --git a/internal/remote/user.go b/internal/remote/user.go
@@ -78,6 +78,18 @@ func (user *User) CloseAndFlushOperationQueue() {
 	user.processWG.Wait()
 }
 
+func (user *User) FinishMailboxIDUpdate(tempID string) error {
+	return user.pushOp(&OpRemMailboxTempID{
+		tempID: tempID,
+	})
+}
+
+func (user *User) FinishMessageIDUpdate(tempID string) error {
+	return user.pushOp(&OpRemMessageTempID{
+		tempID: tempID,
+	})
+}
+
 // CloseAndSerializeOperationQueue closes the remote user.
 func (user *User) CloseAndSerializeOperationQueue() error {
 	ops, err := user.opQueue.closeQueueAndRetrieveRemaining()
diff --git a/internal/remote/user_execute.go b/internal/remote/user_execute.go
@@ -42,6 +42,12 @@ func (user *User) execute(ctx context.Context, op operation) error {
 	case *OpConnMetadataStoreSetValue:
 		return user.executeConnMetadataStoreSetValue(ctx, op)
 
+	case *OpRemMailboxTempID:
+		return user.executeRemMailboxTempID(ctx, op)
+
+	case *OpRemMessageTempID:
+		return user.executeRemMessageTempID(ctx, op)
+
 	default:
 		panic(fmt.Sprintf("bad operation: %v", op))
 	}
@@ -53,19 +59,19 @@ func (user *User) executeMailboxCreate(ctx context.Context, op *OpMailboxCreate)
 		return err
 	}
 
-	user.opQueue.setMailboxID(op.TempID, mbox.ID)
+	user.opQueue.addMailboxTempID(op.TempID, mbox.ID)
 
 	user.send(imap.NewMailboxIDChanged(op.TempID, mbox.ID), true)
 
 	return nil
 }
 
 func (user *User) executeMailboxDelete(ctx context.Context, op *OpMailboxDelete) error {
-	return user.conn.DeleteLabel(ctx, op.MBoxID)
+	return user.conn.DeleteLabel(ctx, user.opQueue.translateMailboxID(op.MBoxID))
 }
 
 func (user *User) executeMailboxUpdate(ctx context.Context, op *OpMailboxUpdate) error {
-	return user.conn.UpdateLabel(ctx, op.MBoxID, op.Name)
+	return user.conn.UpdateLabel(ctx, user.opQueue.translateMailboxID(op.MBoxID), op.Name)
 }
 
 func (user *User) executeMessageCreate(ctx context.Context, op *OpMessageCreate) error {
@@ -74,39 +80,42 @@ func (user *User) executeMessageCreate(ctx context.Context, op *OpMessageCreate)
 		return err
 	}
 
-	user.opQueue.setMessageID(op.TempID, msg.ID)
+	user.opQueue.addMessageTempID(op.TempID, msg.ID)
 
 	user.send(imap.NewMessageIDChanged(op.TempID, msg.ID), true)
 
 	return nil
 }
 
 func (user *User) executeMessageAdd(ctx context.Context, op *OpMessageAdd) error {
-	if err := user.conn.LabelMessages(ctx, op.MessageIDs, op.MBoxID); err != nil {
-		return user.refresh(ctx, op.MessageIDs, op.MBoxID)
+	if err := user.conn.LabelMessages(ctx, user.opQueue.translateMessageIDs(op.MessageIDs...),
+		user.opQueue.translateMailboxID(op.MBoxID)); err != nil {
+		return user.refresh(ctx, user.opQueue.translateMessageIDs(op.MessageIDs...),
+			user.opQueue.translateMailboxID(op.MBoxID))
 	}
 
 	return nil
 }
 
 func (user *User) executeMessageRemove(ctx context.Context, op *OpMessageRemove) error {
-	if err := user.conn.UnlabelMessages(ctx, op.MessageIDs, op.MBoxID); err != nil {
+	if err := user.conn.UnlabelMessages(ctx, user.opQueue.translateMessageIDs(op.MessageIDs...),
+		user.opQueue.translateMailboxID(op.MBoxID)); err != nil {
 		return user.refresh(ctx, op.MessageIDs, op.MBoxID)
 	}
 
 	return nil
 }
 
 func (user *User) executeMessageSeen(ctx context.Context, op *OpMessageSeen) error {
-	if err := user.conn.MarkMessagesSeen(ctx, op.MessageIDs, op.Seen); err != nil {
+	if err := user.conn.MarkMessagesSeen(ctx, user.opQueue.translateMessageIDs(op.MessageIDs...), op.Seen); err != nil {
 		return user.refresh(ctx, op.MessageIDs)
 	}
 
 	return nil
 }
 
 func (user *User) executeMessageFlagged(ctx context.Context, op *OpMessageFlagged) error {
-	if err := user.conn.MarkMessagesFlagged(ctx, op.MessageIDs, op.Flagged); err != nil {
+	if err := user.conn.MarkMessagesFlagged(ctx, user.opQueue.translateMessageIDs(op.MessageIDs...), op.Flagged); err != nil {
 		return user.refresh(ctx, op.MessageIDs)
 	}
 
@@ -162,3 +171,15 @@ func (user *User) executeConnMetadataStoreSetValue(ctx context.Context, op *OpCo
 
 	return nil
 }
+
+func (user *User) executeRemMailboxTempID(ctx context.Context, op *OpRemMailboxTempID) error {
+	user.opQueue.remMailboxTempID(op.tempID)
+
+	return nil
+}
+
+func (user *User) executeRemMessageTempID(ctx context.Context, op *OpRemMessageTempID) error {
+	user.opQueue.remMessageTempID(op.tempID)
+
+	return nil
+}
diff --git a/internal/remote/user_queue.go b/internal/remote/user_queue.go

-Original file line number
+Diff line change
 	github.com/golang/mock v1.6.0
 	github.com/google/uuid v1.3.0
 	github.com/mattn/go-sqlite3 v1.14.10
 -	github.com/pkg/errors v0.9.1
 	github.com/sirupsen/logrus v1.8.1
 	github.com/stretchr/testify v1.7.1-0.20210427113832-6241f9ab9942
 	golang.org/x/exp v0.0.0-20220518171630-0b5c67f07fdf
 	github.com/hashicorp/hcl/v2 v2.10.0 // indirect
 	github.com/kr/pretty v0.3.0 // indirect
 	github.com/mitchellh/go-wordwrap v0.0.0-20150314170334-ad45545899c7 // indirect
 +	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rogpeppe/go-internal v1.8.0 // indirect
 	github.com/zclconf/go-cty v1.8.0 // indirect
Original file line number	Diff line number	Diff line change
`@@ -110,6 +110,10 @@ func (user user) applyMailboxIDChanged(ctx context.Context, tx ent.Tx, update`
`110`	`110`	`return err`
`111`	`111`	`}`
`112`	`112`
	`113`	`+ if err := user.remote.FinishMailboxIDUpdate(update.OldID); err != nil {`
	`114`	`+ logrus.WithError(err).Errorf("Call to FinishMailboxIDUpdate() failed")`
	`115`	`+ }`
	`116`	`+`
`113`	`117`	`return txUpdateMailboxID(ctx, tx, update.OldID, update.NewID)`
`114`	`118`	`}`
`115`	`119`
`@@ -208,6 +212,10 @@ func (user user) applyMessageIDChanged(ctx context.Context, tx ent.Tx, update`
`208`	`212`	`return err`
`209`	`213`	`}`
`210`	`214`
	`215`	`+ if err := user.remote.FinishMessageIDUpdate(update.OldID); err != nil {`
	`216`	`+ logrus.WithError(err).Error("Call to FinishMessageIDUpdate() failed")`
	`217`	`+ }`
	`218`	`+`
`211`	`219`	`if err := user.store.Update(update.OldID, update.NewID); err != nil {`
`212`	`220`	`return err`
`213`	`221`	`}`
Original file line number	Diff line number	Diff line change
`@@ -28,12 +28,4 @@ func (op *OpMessageFlagged) merge(other operation) (operation, bool) {`
`28`	`28`	`}`
`29`	`29`	`}`
`30`	`30`
`31`		`-func (op *OpMessageFlagged) setMessageID(tempID, messageID string) {`
`32`		`- for idx := range op.MessageIDs {`
`33`		`- if op.MessageIDs[idx] == tempID {`
`34`		`- op.MessageIDs[idx] = messageID`
`35`		`- }`
`36`		`- }`
`37`		`-}`
`38`		`-`
`39`	`31`	`func (OpMessageFlagged) _isOperation() {}`
Original file line number	Diff line number	Diff line change
`@@ -28,12 +28,4 @@ func (op *OpMessageSeen) merge(other operation) (operation, bool) {`
`28`	`28`	`}`
`29`	`29`	`}`
`30`	`30`
`31`		`-func (op *OpMessageSeen) setMessageID(tempID, messageID string) {`
`32`		`- for idx := range op.MessageIDs {`
`33`		`- if op.MessageIDs[idx] == tempID {`
`34`		`- op.MessageIDs[idx] = messageID`
`35`		`- }`
`36`		`- }`
`37`		`-}`
`38`		`-`
`39`	`31`	`func (OpMessageSeen) _isOperation() {}`