chore_: remove message ids from query queue after ack

Author: kaichaosun

eth-node/bridge/geth/waku.go

@@ -314,3 +314,6 @@ func GetWakuFilterFrom(f types.Filter) *wakucommon.Filter {
 func (w *wakuFilterWrapper) ID() string {
 	return w.id
 }
+
+func (w *GethWakuWrapper) ConfirmMessageDelivered(hashes []common.Hash) {
+}

eth-node/bridge/geth/wakuv2.go

@@ -327,3 +327,7 @@ func GetWakuV2FilterFrom(f types.Filter) *wakucommon.Filter {
 func (w *wakuV2FilterWrapper) ID() string {
 	return w.id
 }
+
+func (w *gethWakuV2Wrapper) ConfirmMessageDelivered(hashes []common.Hash) {
+	w.waku.ConfirmMessageDelivered(hashes)
+}

eth-node/types/waku.go

@@ -178,4 +178,7 @@ type Waku interface {
 
 	// ClearEnvelopesCache clears waku envelopes cache
 	ClearEnvelopesCache()
+
+	// ConfirmMessageDelivered updates a message has been delivered in waku
+	ConfirmMessageDelivered(hash []common.Hash)
 }

protocol/messenger_peersyncing.go

@@ -43,6 +43,8 @@ func (m *Messenger) markDeliveredMessages(acks [][]byte) {
 			m.logger.Debug("Can't set message status as delivered", zap.Error(err))
 		}
 
+		m.transport.ConfirmMessageDelivered(messageID)
+
 		//send signal to client that message status updated
 		if m.config.messengerSignalsHandler != nil {
 			message, err := m.persistence.MessageByID(messageID)

protocol/transport/transport.go

@@ -717,3 +717,15 @@ func (t *Transport) RemovePubsubTopicKey(topic string) error {
 	}
 	return nil
 }
+
+func (t *Transport) ConfirmMessageDelivered(messageID string) {
+	hashes, ok := t.envelopesMonitor.identifierHashes[messageID]
+	if !ok {
+		return
+	}
+	commHashes := make([]common.Hash, len(hashes))
+	for i, h := range hashes {
+		commHashes[i] = common.BytesToHash(h[:])
+	}
+	t.waku.ConfirmMessageDelivered(commHashes)
+}

wakuv2/waku.go

@@ -1042,6 +1042,21 @@ func (w *Waku) checkIfMessagesStored() {
 	}
 }
 
+func (w *Waku) ConfirmMessageDelivered(hashes []gethcommon.Hash) {
+	w.sendMsgIDsMu.Lock()
+	defer w.sendMsgIDsMu.Unlock()
+	for pubsubTopic, subMsgs := range w.sendMsgIDs {
+		for _, hash := range hashes {
+			delete(subMsgs, hash)
+			if len(subMsgs) == 0 {
+				delete(w.sendMsgIDs, pubsubTopic)
+			} else {
+				w.sendMsgIDs[pubsubTopic] = subMsgs
+			}
+		}
+	}
+}
+
 type publishFn = func(envelope *protocol.Envelope, logger *zap.Logger) error
 
 func (w *Waku) publishEnvelope(envelope *protocol.Envelope, publishFn publishFn, logger *zap.Logger) {