Try to recover from corrupted NATS streams in memory temporarily (#2301)

1 files changed, 17 insertions, 0 deletions

diff --git a/setup/process/process.go b/setup/process/process.go
index d55751d7..01eb26e2 100644
--- a/setup/process/process.go
+++ b/setup/process/process.go
@@ -2,13 +2,19 @@ package process
 
 import (
 	"context"
+	"fmt"
 	"sync"
+
+	"github.com/getsentry/sentry-go"
+	"github.com/sirupsen/logrus"
+	"go.uber.org/atomic"
 )
 
 type ProcessContext struct {
 	wg       *sync.WaitGroup    // used to wait for components to shutdown
 	ctx      context.Context    // cancelled when Stop is called
 	shutdown context.CancelFunc // shut down Dendrite
+	degraded atomic.Bool
 }
 
 func NewProcessContext() *ProcessContext {
@@ -43,3 +49,14 @@ func (b *ProcessContext) WaitForShutdown() <-chan struct{} {
 func (b *ProcessContext) WaitForComponentsToFinish() {
 	b.wg.Wait()
 }
+
+func (b *ProcessContext) Degraded() {
+	if b.degraded.CAS(false, true) {
+		logrus.Warn("Dendrite is running in a degraded state")
+		sentry.CaptureException(fmt.Errorf("Process is running in a degraded state"))
+	}
+}
+
+func (b *ProcessContext) IsDegraded() bool {
+	return b.degraded.Load()
+}