implemented garbled message handling

context/params/message.go

 package params
 
+import "time"
+
 type Messages struct {
 	MessageReceptionBuffLen        uint
 	MessageReceptionWorkerPoolSize uint
+	MaxChecksGarbledMessage        uint
+	GarbledMessageWait             time.Duration
 }
 
 func GetDefaultMessage() Messages {
 	return Messages{
 		MessageReceptionBuffLen:        500,
 		MessageReceptionWorkerPoolSize: 4,
+		MaxChecksGarbledMessage:        10,
+		GarbledMessageWait:             15 * time.Minute,
 	}
 }

network/keyExchange/exchange.go

@@ -12,7 +12,7 @@ const keyExchangeTriggerName = "KeyExchangeTrigger"
 const keyExchangeConfirmName = "KeyExchangeConfirm"
 const keyExchangeMulti = "KeyExchange"
 
-func Start(ctx *context.Context) stoppable.Stoppable {
+func Start(ctx *context.Context, garbledMessageTrigger chan<- struct{}) stoppable.Stoppable {
 
 	// register the rekey trigger thread
 	triggerCh := make(chan message.Receive, 100)
@@ -28,7 +28,7 @@ func Start(ctx *context.Context) stoppable.Stoppable {
 		})
 
 	// start the trigger thread
-	go startTrigger(ctx, triggerCh, triggerStop)
+	go startTrigger(ctx, triggerCh, triggerStop, garbledMessageTrigger)
 
 	//register the rekey confirm thread
 	confirmCh := make(chan message.Receive, 100)

network/keyExchange/trigger.go

@@ -23,18 +23,23 @@ const (
 )
 
 func startTrigger(ctx *context.Context, c chan message.Receive,
-	stop *stoppable.Single) {
+	stop *stoppable.Single, garbledMessageTrigger chan<- struct{}) {
 	for true {
 		select {
 		case <-stop.Quit():
 			return
 		case request := <-c:
-			handleTrigger(ctx, request)
+			err := handleTrigger(ctx, request, garbledMessageTrigger)
+			if err != nil {
+				jww.ERROR.Printf("Failed to handle rekey trigger: %s",
+					err)
+			}
 		}
 	}
 }
 
-func handleTrigger(ctx *context.Context, request message.Receive) error {
+func handleTrigger(ctx *context.Context, request message.Receive,
+	garbledMessageTrigger chan<- struct{}) error {
 	//ensure the message was encrypted properly
 	if request.Encryption != message.E2E {
 		errMsg := fmt.Sprintf(errBadTrigger, request.Sender)
@@ -78,6 +83,14 @@ func handleTrigger(ctx *context.Context, request message.Receive) error {
 		jww.INFO.Printf("New session from Key Exchange Trigger to "+
 			"create session %s for partner %s is a duplicate, request ignored",
 			session.GetID(), request.Sender)
+	} else {
+		//if the session is new, attempt to trigger garbled message processing
+		//if there is contention, skip
+		select {
+		case garbledMessageTrigger <- struct{}{}:
+		default:
+			jww.WARN.Println("Failed to trigger garbled messages")
+		}
 	}
 
 	//Send the Confirmation Message

network/manager.go

@@ -139,7 +139,7 @@ func (m *manager) StartRunners() error {
 	m.runners.Add(m.round.StartProcessors())
 
 	// Key exchange
-	m.runners.Add(keyExchange.Start(m.Context))
+	m.runners.Add(keyExchange.Start(m.Context, m.message.GetTriggerGarbledCheckChannel()))
 
 	return nil
 }

network/message/garbled.go

+package message
+
+import (
+	"gitlab.com/elixxir/client/context/message"
+	"time"
+)
+
+func (m *Manager) processGarbledMessages(quitCh <-chan struct{}) {
+	done := false
+	for !done {
+		select {
+		case <-quitCh:
+			done = true
+		case <-m.triggerGarbled:
+			m.handleGarbledMessages()
+		}
+	}
+}
+
+func (m *Manager) handleGarbledMessages() {
+	garbledMsgs := m.Session.GetGarbledMessages()
+	e2eKv := m.Session.E2e()
+	//try to decrypt every garbled message, excising those who's counts are too high
+	for grbldMsg, count, timestamp, has := garbledMsgs.Next(); has;
+	grbldMsg, count, timestamp, has = garbledMsgs.Next() {
+		fingerprint := grbldMsg.GetKeyFP()
+		// Check if the key is there, process it if it is
+		if key, isE2E := e2eKv.PopKey(fingerprint); isE2E {
+			// Decrypt encrypted message
+			msg, err := key.Decrypt(grbldMsg)
+			// get the sender
+			sender := key.GetSession().GetPartner()
+			if err == nil {
+				//remove from the buffer if decryption is successful
+				garbledMsgs.Remove(grbldMsg)
+				//handle the successfully decrypted message
+				xxMsg, ok := m.partitioner.HandlePartition(sender, message.E2E, msg.GetContents())
+				if ok {
+					m.Switchboard.Speak(xxMsg)
+					continue
+				}
+			}
+		}
+		// fail the message if any part of the decryption fails,
+		// unless it is our of attempts and has been in the buffer long enough,
+		// then remove it
+		if count == m.param.MaxChecksGarbledMessage &&
+			time.Since(timestamp) > m.param.GarbledMessageWait {
+			garbledMsgs.Remove(grbldMsg)
+		} else {
+			garbledMsgs.Failed(grbldMsg)
+		}
+	}
+}

network/message/manager.go

@@ -18,6 +18,7 @@ type Manager struct {
 	messageReception chan Bundle
 	nodeRegistration chan network.NodeGateway
 	networkIsHealthy chan bool
+	triggerGarbled   chan struct{}
 }
 
 func NewManager(internal internal.Internal, param params.Messages,
@@ -28,6 +29,7 @@ func NewManager(internal internal.Internal, param params.Messages,
 		partitioner:      parse.NewPartitioner(dummyMessage.ContentsSize(), internal.Session),
 		messageReception: make(chan Bundle, param.MessageReceptionBuffLen),
 		networkIsHealthy: make(chan bool, 1),
+		triggerGarbled:   make(chan struct{}, 1),
 		nodeRegistration: nodeRegistration,
 	}
 	m.Internal = internal
@@ -39,19 +41,33 @@ func (m *Manager) GetMessageReceptionChannel() chan<- Bundle {
 	return m.messageReception
 }
 
+//Gets the channel to send received messages on
+func (m *Manager) GetTriggerGarbledCheckChannel() chan<- struct{} {
+	return m.triggerGarbled
+}
+
 //Starts all worker pool
 func (m *Manager) StartProcessies() stoppable.Stoppable {
 	multi := stoppable.NewMulti("MessageReception")
 
+	//create the message reception workers
 	for i := uint(0); i < m.param.MessageReceptionWorkerPoolSize; i++ {
 		stop := stoppable.NewSingle(fmt.Sprintf("MessageReception Worker %v", i))
 		go m.processMessages(stop.Quit())
 		multi.Add(stop)
 	}
 
-	critStop := stoppable.NewSingle("Critical Messages Handler")
+	//create the critical messages thread
+	critStop := stoppable.NewSingle("CriticalMessages")
 	go m.processCriticalMessages(critStop.Quit())
 	m.Health.AddChannel(m.networkIsHealthy)
+	multi.Add(critStop)
+
+	//create the garbled messages thread
+	garbledStop := stoppable.NewSingle("GarbledMessages")
+	go m.processGarbledMessages(garbledStop.Quit())
+	multi.Add(garbledStop)
+
 
 	return multi
 }

storage/utility/messageBuffer.go

@@ -261,7 +261,7 @@ func next(msgMap map[MessageHash]struct{}) MessageHash {
 	return MessageHash{}
 }
 
-// Succeeded sets a messaged as processed and removed it from the buffer.
+// Remove sets a messaged as processed and removed it from the buffer.
 func (mb *MessageBuffer) Succeeded(m interface{}) {
 	h := mb.handler.HashMessage(m)
 

storage/utility/messageBuffer_test.go

@@ -233,7 +233,7 @@ func TestMessageBuffer_Next(t *testing.T) {
 	}
 }
 
-// Tests happy path of MessageBuffer.Succeeded().
+// Tests happy path of MessageBuffer.Remove().
 func TestMessageBuffer_Succeeded(t *testing.T) {
 	th := newTestHandler()
 	// Create new MessageBuffer and fill with message
@@ -254,7 +254,7 @@ func TestMessageBuffer_Succeeded(t *testing.T) {
 	_, exists1 := testMB.messages[th.HashMessage(m)]
 	_, exists2 := testMB.processingMessages[th.HashMessage(m)]
 	if exists1 || exists2 {
-		t.Errorf("Succeeded() did not remove the message from the buffer."+
+		t.Errorf("Remove() did not remove the message from the buffer."+
 			"\n\tbuffer: %+v", testMB)
 	}
 }

storage/utility/meteredCmixMessageBuffer.go

@@ -142,7 +142,7 @@ func (mcmb *MeteredCmixMessageBuffer) Next() (format.Message, uint, time.Time, b
 	return msfFormat, rtnCnt, msg.Timestamp, true
 }
 
-func (mcmb *MeteredCmixMessageBuffer) Succeeded(m format.Message) {
+func (mcmb *MeteredCmixMessageBuffer) Remove(m format.Message) {
 	mcmb.mb.Succeeded(m)
 }