From 38a60a69df968398ee04aee651e400aec902525c Mon Sep 17 00:00:00 2001
From: Michael Quigley <michael@quigley.com>
Date: Thu, 11 Feb 2021 13:16:18 -0500
Subject: [PATCH] Wired in the fault reporting. (#184)

---
 router/forwarder/faulter.go    |  4 ++++
 router/forwarder/forwarder.go  |  4 ++++
 router/router.go               |  2 +-
 router/xgress/retransmitter.go | 12 +++++++++---
 4 files changed, 18 insertions(+), 4 deletions(-)

diff --git a/router/forwarder/faulter.go b/router/forwarder/faulter.go
index cf613946e..96b12a76f 100644
--- a/router/forwarder/faulter.go
+++ b/router/forwarder/faulter.go
@@ -33,6 +33,10 @@ func newFaulter(interval time.Duration) *faulter {
 	return f
 }
 
+func (self *faulter) report(sessionId string) {
+	self.sessionIds.Set(sessionId, struct{}{})
+}
+
 func (self *faulter) run() {
 	logrus.Infof("started")
 	defer logrus.Errorf("exited")
diff --git a/router/forwarder/forwarder.go b/router/forwarder/forwarder.go
index 186c39c3d..1e9ce450c 100644
--- a/router/forwarder/forwarder.go
+++ b/router/forwarder/forwarder.go
@@ -183,6 +183,10 @@ func (forwarder *Forwarder) ForwardAcknowledgement(srcAddr xgress.Address, ackno
 	}
 }
 
+func (forwarder *Forwarder) ForwardingFault(sessionId string) {
+	forwarder.faulter.report(sessionId)
+}
+
 func (forwarder *Forwarder) Debug() string {
 	return forwarder.sessions.debug() + forwarder.destinations.debug()
 }
diff --git a/router/router.go b/router/router.go
index e6e76b62d..1927084bc 100644
--- a/router/router.go
+++ b/router/router.go
@@ -86,7 +86,7 @@ func Create(config *Config, versionProvider common.VersionProvider) *Router {
 
 	xgress.InitPayloadIngester(closeNotify)
 	xgress.InitAcker(fwd, metricsRegistry, closeNotify)
-	xgress.InitRetransmitter(fwd, metricsRegistry, closeNotify)
+	xgress.InitRetransmitter(fwd, fwd, metricsRegistry, closeNotify)
 
 	return &Router{
 		config:          config,
diff --git a/router/xgress/retransmitter.go b/router/xgress/retransmitter.go
index f91b7cdaa..0c1b0df2f 100644
--- a/router/xgress/retransmitter.go
+++ b/router/xgress/retransmitter.go
@@ -8,12 +8,17 @@ import (
 
 var retransmitter *Retransmitter
 
-func InitRetransmitter(forwarder PayloadBufferForwarder, metrics metrics.Registry, closeNotify <-chan struct{}) {
-	retransmitter = NewRetransmitter(forwarder, metrics, closeNotify)
+func InitRetransmitter(forwarder PayloadBufferForwarder, faultReporter RetransmitterFaultReporter, metrics metrics.Registry, closeNotify <-chan struct{}) {
+	retransmitter = NewRetransmitter(forwarder, faultReporter, metrics, closeNotify)
+}
+
+type RetransmitterFaultReporter interface {
+	ForwardingFault(sessionId string)
 }
 
 type Retransmitter struct {
 	forwarder            PayloadBufferForwarder
+	faultReporter        RetransmitterFaultReporter
 	retxTail             *txPayload
 	retxHead             *txPayload
 	retransmitIngest     chan *txPayload
@@ -22,7 +27,7 @@ type Retransmitter struct {
 	closeNotify          <-chan struct{}
 }
 
-func NewRetransmitter(forwarder PayloadBufferForwarder, metrics metrics.Registry, closeNotify <-chan struct{}) *Retransmitter {
+func NewRetransmitter(forwarder PayloadBufferForwarder, faultReporter RetransmitterFaultReporter, metrics metrics.Registry, closeNotify <-chan struct{}) *Retransmitter {
 	ctrl := &Retransmitter{
 		forwarder:        forwarder,
 		retransmitIngest: make(chan *txPayload, 16),
@@ -144,6 +149,7 @@ func (retransmitter *Retransmitter) retransmitSender() {
 					if !retransmit.x.closed.Get() {
 						logger.WithError(err).Errorf("unexpected error while retransmitting payload from [@/%v]", retransmit.x.address)
 						retransmissionFailures.Mark(1)
+						retransmitter.faultReporter.ForwardingFault(retransmit.payload.SessionId)
 					} else {
 						logger.WithError(err).Tracef("unexpected error while retransmitting payload from [@/%v] (already closed)", retransmit.x.address)
 					}