fix: re-enqueue message when we fail to decrypt it

Signed-off-by: gfanton <8671905+gfanton@users.noreply.github.com>
berty · Mar 31, 2022 · 9bc63f3 · 9bc63f3
1 parent 24e19e1
commit 9bc63f3
Show file tree

Hide file tree

Showing 4 changed files with 21 additions and 8 deletions.
diff --git a/go/pkg/bertyprotocol/group.go b/go/pkg/bertyprotocol/group.go
@@ -151,7 +151,7 @@ func FillMessageKeysHolderUsingNewData(ctx context.Context, gc *GroupContext) <-
 
 			// A new chainKey is registered, check if cached messages can be opened with it
 			if rawPK, err := pk.Raw(); err == nil {
-				gc.MessageStore().ProcessMessageQueueForDevicePK(rawPK)
+				gc.MessageStore().ProcessMessageQueueForDevicePK(ctx, rawPK)
 			}
 
 			ch <- pk

diff --git a/go/pkg/bertyprotocol/store_message.go b/go/pkg/bertyprotocol/store_message.go
@@ -119,13 +119,16 @@ func (m *MessageStore) CacheSizeForDevicePK(devicePK []byte) (size int, ok bool)
 	return
 }
 
-func (m *MessageStore) ProcessMessageQueueForDevicePK(devicePK []byte) {
+func (m *MessageStore) ProcessMessageQueueForDevicePK(ctx context.Context, devicePK []byte) {
 	m.muDeviceCaches.Lock()
 	device, ok := m.deviceCaches[string(devicePK)]
 	if ok {
-		device.hasSecret = true
-		if next := device.queue.Next(); next != nil {
-			m.cmessage <- next
+		if device.hasSecret = m.mks.HasSecretForRawDevicePK(ctx, m.g.PublicKey, devicePK); device.hasSecret {
+			if next := device.queue.Next(); next != nil {
+				m.cmessage <- next
+			}
+		} else {
+			m.logger.Error("unable to process message, no secret found for device pk", logutil.PrivateBinary("devicepk", devicePK))
 		}
 	}
 	m.muDeviceCaches.Unlock()
@@ -209,8 +212,16 @@ func (m *MessageStore) processMessageLoop(ctx context.Context) {
 				}
 
 				if err := m.processMessage(ctx, ownPK, message); err != nil {
-					m.logger.Error("unable to process message", zap.Error(err))
-					return
+					if errcode.Is(err, errcode.ErrCryptoDecryptPayload) {
+						// @FIXME(gfanton): this should not happen
+						m.logger.Warn("unable to open envelope, adding envelope to cache for later process", zap.Error(err))
+
+						// if failed to decrypt add to queue, for later process
+						device.queue.Add(message)
+						_ = m.emitters.groupCacheMessage.Emit(*message)
+					} else {
+						m.logger.Error("unable to prcess message", zap.Error(err))
+					}
 				} else if next := device.queue.Next(); next != nil {
 					m.cmessage <- next
 				}

diff --git a/go/pkg/bertyprotocol/store_message_test.go b/go/pkg/bertyprotocol/store_message_test.go
@@ -195,7 +195,7 @@ func Test_Add_Messages_To_Cache(t *testing.T) {
 		new(protocoltypes.GroupMessageEvent), eventbus.BufSize(entriesCount))
 	require.NoError(t, err)
 
-	peers[1].GC.MessageStore().ProcessMessageQueueForDevicePK(dPK0Raw)
+	peers[1].GC.MessageStore().ProcessMessageQueueForDevicePK(ctx, dPK0Raw)
 
 	// check that all events has been received on peer 2
 	for i := 0; i < entriesCount; i++ {

diff --git a/go/pkg/bertyreplication/services_replication_test.go b/go/pkg/bertyreplication/services_replication_test.go
@@ -349,6 +349,8 @@ func TestReplicationService_ReplicateGroupStats_ReplicateGlobalStats(t *testing.
 }
 
 func TestReplicationService_Flow(t *testing.T) {
+	testutil.FilterSpeed(t, testutil.Slow)
+
 	logger, cleanup := testutil.Logger(t)
 	defer cleanup()
 	ctx, cancel := context.WithCancel(context.Background())