[raft] Improve /healthy status to return an error when raft errored

Author: the-glu

cmds/core-service/main.go

@@ -24,6 +24,7 @@ import (
 	auxs "github.com/interuss/dss/pkg/aux_/store"
 	"github.com/interuss/dss/pkg/build"
 	"github.com/interuss/dss/pkg/logging"
+	"github.com/interuss/dss/pkg/raftstore"
 	"github.com/interuss/dss/pkg/rid/application"
 	rid_v1 "github.com/interuss/dss/pkg/rid/server/v1"
 	rid_v2 "github.com/interuss/dss/pkg/rid/server/v2"
@@ -399,6 +400,15 @@ func RunHTTPServer(ctx context.Context, ctxCanceler func(), address, locality st
 func healthyEndpointMiddleware(logger *zap.Logger, next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if r.URL.Path == "/healthy" {
+
+			if params.GetStoreParameters().StoreType == params.RaftStoreType && !raftstore.IsHealthy() {
+				w.WriteHeader(http.StatusServiceUnavailable)
+				if _, err := w.Write([]byte("unhealthy: raft consensus is down")); err != nil {
+					logger.Error("Error writing to /healthy")
+				}
+				return
+			}
+
 			if _, err := w.Write([]byte("ok")); err != nil {
 				logger.Error("Error writing to /healthy")
 			}

pkg/raftstore/consensus/consensus.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"net/url"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/interuss/dss/pkg/logging"
@@ -42,6 +43,10 @@ type Consensus struct {
 	confState     raftpb.ConfState
 	snapshotIndex uint64
 	appliedIndex  uint64
+
+	// failed is set once handleReady exits, meaning this node's consensus
+	// loop has stopped running and it can no longer process or apply entries.
+	failed atomic.Bool
 }
 
 func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params.ConnectParameters, provider snapshotProvider, commitC chan<- EntryCommit) (*Consensus, error) {
@@ -103,7 +108,8 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params.
 			consensus.logger.Error("handleReady exited with error, shutting down consensus", zap.Error(err))
 		}
 
-		consensus.Stop(context.Background())
+		consensus.failed.Store(true)
+		consensus.Stop(ctx)
 	}()
 
 	return consensus, nil
@@ -126,6 +132,16 @@ func (c *Consensus) Stop(ctx context.Context) {
 	})
 }
 
+// IsHealthy reports whether this node's consensus loop is still running and
+// the cluster currently has an elected leader. A node with no known leader
+// has lost quorum (or hasn't joined yet) and cannot serve requests.
+func (c *Consensus) IsHealthy() bool {
+	if c.failed.Load() {
+		return false
+	}
+	return c.node.Status().Lead != 0
+}
+
 // ProposeValue blocks until the proposal is committed and applied / dropped or until ctx is cancelled.
 func (c *Consensus) ProposeValue(ctx context.Context, requestType string, payload any, readOnly bool) (any, error) {
 	proposal, err := newProposal(ctx, requestType, payload, readOnly)

pkg/raftstore/store.go

@@ -95,3 +95,12 @@ func (s *Store[R]) processCommits(ctx context.Context, commitCh <-chan consensus
 		}
 	}
 }
+
+// IsHealthy reports whether the shared raft consensus instance is up and has
+// an elected leader. Returns false if no raftstore has been initialized yet.
+func IsHealthy() bool {
+	if sharedConsensus == nil {
+		return false
+	}
+	return sharedConsensus.IsHealthy()
+}