diff --git a/cmd/pad/cmd_server.go b/cmd/pad/cmd_server.go index efa4b3001..04313dbde 100644 --- a/cmd/pad/cmd_server.go +++ b/cmd/pad/cmd_server.go @@ -648,6 +648,16 @@ func serveCmd() *cobra.Command { if err := rc.Ping(context.Background()).Err(); err != nil { return fmt.Errorf("redis connection failed: %w", err) } + // Route go-redis's own diagnostics into slog (BUG-2727). + // Its default logger writes to the standard log package, + // so its messages bypass the structured pipeline entirely + // — including the one that matters most here: "channel is + // full ... message is dropped", emitted when a + // subscription's 100-deep buffer stays full past the 60s + // send timeout. That is a silent notification loss whose + // only trace was a line nobody's log aggregator was + // shaped to catch. + redis.SetLogger(redisSlogLogger{}) eventBus = events.NewRedisBus(rc) watchRedis = rc slog.Info("Event bus using Redis pub/sub", "addr", opts.Addr, "db", opts.DB) @@ -671,10 +681,20 @@ func serveCmd() *cobra.Command { // doc for what each implementation does and does not fix. var watchBus watchevents.Bus if watchRedis != nil { - watchBus = watchevents.NewRedisBus(watchRedis) + redisWatchBus := watchevents.NewRedisBus(watchRedis) + // Operational instrumentation (BUG-2727). Attached to the + // concrete type because the conditions it reports — + // dropped notifications, sequence gaps, id-space resets, + // the receive loop stopping — are detected inside the bus + // and are not visible at the Bus interface, so a wrapper + // of the events.EventBus kind cannot see them. + redisWatchBus.SetObserver(metrics.NewWatchEventsObserver(m)) + watchBus = redisWatchBus slog.Info("Watch notification bus using Redis pub/sub") } else { - watchBus = watchevents.New() + memWatchBus := watchevents.New() + memWatchBus.SetObserver(metrics.NewWatchEventsObserver(m)) + watchBus = memWatchBus slog.Info("Watch notification bus using in-memory (single instance)") } srv.SetWatchEventsBus(watchBus) @@ -697,6 +717,13 @@ func serveCmd() *cobra.Command { var redisPresence *server.RedisSessionPresence if watchRedis != nil { redisPresence = server.NewRedisSessionPresence(watchRedis) + // Presence is fail-soft by design — a failed write leaves + // a live session unlisted rather than dropping its + // connection — so its failures have no user-visible + // signal beyond a push that quietly reaches fewer + // sessions. The counter is the only alertable trace + // (BUG-2727). + redisPresence.SetObserver(server.NewMetricsPresenceObserver(m)) // Backstop for early-return paths that never reach the // shutdown sequence; Close is idempotent. Deliberately does // NOT delete this instance's entries — a shutdown racing a @@ -711,6 +738,24 @@ func serveCmd() *cobra.Command { } srv.SetSessionPresence(sessionPresence) + // Redis reachability prober (BUG-2727). Reports into + // /health/ready's payload and pad_redis_up; deliberately does + // NOT gate readiness — see server.RedisHealth's doc comment. + // Only wired when there IS a Redis, so a single-process + // deployment reports no redis block rather than a false one. + if watchRedis != nil { + redisHealth := server.NewRedisHealth(watchRedis, func(ok bool) { + if ok { + m.RedisUp.Set(1) + } else { + m.RedisUp.Set(0) + } + }) + redisHealth.Start() + defer redisHealth.Stop() + srv.SetRedisHealth(redisHealth) + } + // Yjs collab room manager (PLAN-1248). Single-instance only // today; multi-replica fanout via Redis is a deferred IDEA. // MemoryOpBus is in-process; the OpBus interface keeps the diff --git a/cmd/pad/redis_logger.go b/cmd/pad/redis_logger.go new file mode 100644 index 000000000..4f0dd668d --- /dev/null +++ b/cmd/pad/redis_logger.go @@ -0,0 +1,30 @@ +package main + +import ( + "context" + "fmt" + "log/slog" +) + +// redisSlogLogger adapts go-redis's internal logging interface onto slog +// (BUG-2727). +// +// go-redis logs a handful of operational conditions through a package-level +// logger that defaults to the standard `log` package. Two of them are +// notification loss and neither has any other trace: +// +// - "channel is full for (message is dropped)" — a subscription's +// buffer stayed full past the send timeout, so go-redis discarded a +// message before Pad ever saw it. Pad detects the CONSEQUENCE as a +// sequence gap (see watchevents.Observer), but the cause is only visible +// here. +// - "unknown message type" — a protocol-level surprise on the subscription. +// +// Logged at WARN rather than ERROR: none of them is actionable on its own, +// and go-redis also logs benign reconnect chatter through the same call. +// The metrics are what to alert on; this is what to read afterwards. +type redisSlogLogger struct{} + +func (redisSlogLogger) Printf(_ context.Context, format string, v ...interface{}) { + slog.Warn("redis: " + fmt.Sprintf(format, v...)) +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 42d03675d..42d38bb57 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -27,6 +27,51 @@ type Metrics struct { EventBusPublishTotal *prometheus.Counter EventBusSubscribers *prometheus.Gauge + // Redis operability metrics (BUG-2727). Wired from + // cmd/pad/cmd_server.go and only meaningful on a deployment with + // PAD_REDIS_URL set — a single-process binary never touches Redis, so + // these stay at their zero values there, which is the honest reading + // rather than an absence. + // + // RedisUp is written by internal/server's health prober, NOT sampled + // on scrape: a collector that dials on every scrape turns a monitoring + // system into a load generator and makes the metric's value depend on + // who is asking. It is 1 when the last probe succeeded, 0 when it + // failed, and absent entirely when the deployment has no Redis. + RedisUp prometheus.Gauge + + // WatchNotificationsDroppedTotal counts notifications this instance + // RECEIVED and could not hand to one of its own subscribers. It does + // not count go-redis discarding messages before we see them — that + // shows up as a sequence gap instead; see watchevents.Observer for why + // the distinction matters when reading these two together. + WatchNotificationsDroppedTotal *prometheus.CounterVec + + // WatchSequenceGapsTotal counts gap EVENTS; + // WatchNotificationsMissedTotal counts the notifications those gaps + // span. Both, because one gap of 500 and 500 gaps of one are very + // different incidents and a single counter cannot tell them apart. + WatchSequenceGapsTotal prometheus.Counter + WatchNotificationsMissedTotal prometheus.Counter + + // WatchSequenceResetsTotal counts id-space changes (epoch change or + // the shared counter going backwards) — each one drops this + // instance's replay buffer, so resumes across it answer + // sync_required. + WatchSequenceResetsTotal *prometheus.CounterVec + + // WatchReceiveLoopExitsTotal counts the receive loop stopping. Any + // non-zero value outside a shutdown means an instance that publishes + // fine and receives nothing. + WatchReceiveLoopExitsTotal prometheus.Counter + + // SessionPresenceFailuresTotal counts failed presence operations by + // op. A sustained non-zero rate means sessions are missing from + // GET /api/v1/sessions and untargetable by a session-directed push, + // which is invisible to the user beyond a push that quietly reaches + // nobody. + SessionPresenceFailuresTotal *prometheus.CounterVec + // MCP traffic metrics (PLAN-943 TASK-961). Wired from // internal/server/middleware_mcp_audit.go (per-request seam) and // internal/server/middleware_mcp_auth.go + middleware_auth.go @@ -190,7 +235,49 @@ func New() *Metrics { Buckets: []float64{10, 60, 300, 900, 3600, 21600, 86400, 7 * 86400, 30 * 86400, 60 * 86400}, }) + redisUp := prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "pad_redis_up", + Help: "1 if the last Redis health probe succeeded, 0 if it failed. Not exported when the deployment has no Redis.", + }) + + watchNotificationsDroppedTotal := prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "pad_watchevents_notifications_dropped_total", + Help: "Notifications received by this instance but not delivered to a local subscriber, by reason.", + }, []string{"reason"}) + + watchSequenceGapsTotal := prometheus.NewCounter(prometheus.CounterOpts{ + Name: "pad_watchevents_sequence_gaps_total", + Help: "Times this instance detected a forward jump in the notification id sequence, i.e. it missed at least one notification.", + }) + + watchNotificationsMissedTotal := prometheus.NewCounter(prometheus.CounterOpts{ + Name: "pad_watchevents_notifications_missed_total", + Help: "Notifications spanned by detected sequence gaps — how many were missed, where pad_watchevents_sequence_gaps_total counts how often.", + }) + + watchSequenceResetsTotal := prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "pad_watchevents_sequence_resets_total", + Help: "Times the notification id space changed under this instance, dropping its replay buffer, by reason.", + }, []string{"reason"}) + + watchReceiveLoopExitsTotal := prometheus.NewCounter(prometheus.CounterOpts{ + Name: "pad_watchevents_receive_loop_exits_total", + Help: "Times the Redis subscription receive loop stopped. Non-zero outside shutdown means this instance receives no notifications at all.", + }) + + sessionPresenceFailuresTotal := prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "pad_session_presence_failures_total", + Help: "Failed session-presence operations by op. Failures leave sessions unlisted and untargetable.", + }, []string{"op"}) + reg.MustRegister( + redisUp, + watchNotificationsDroppedTotal, + watchSequenceGapsTotal, + watchNotificationsMissedTotal, + watchSequenceResetsTotal, + watchReceiveLoopExitsTotal, + sessionPresenceFailuresTotal, httpRequestsTotal, httpRequestDuration, httpResponseSize, @@ -208,7 +295,16 @@ func New() *Metrics { ) return &Metrics{ - Registry: reg, + Registry: reg, + + RedisUp: redisUp, + WatchNotificationsDroppedTotal: watchNotificationsDroppedTotal, + WatchSequenceGapsTotal: watchSequenceGapsTotal, + WatchNotificationsMissedTotal: watchNotificationsMissedTotal, + WatchSequenceResetsTotal: watchSequenceResetsTotal, + WatchReceiveLoopExitsTotal: watchReceiveLoopExitsTotal, + SessionPresenceFailuresTotal: sessionPresenceFailuresTotal, + HTTPRequestsTotal: httpRequestsTotal, HTTPRequestDuration: httpRequestDuration, HTTPResponseSize: httpResponseSize, diff --git a/internal/metrics/watchevents_observer.go b/internal/metrics/watchevents_observer.go new file mode 100644 index 000000000..08ae8408a --- /dev/null +++ b/internal/metrics/watchevents_observer.go @@ -0,0 +1,50 @@ +package metrics + +import "github.com/PerpetualSoftware/pad/internal/watchevents" + +// WatchEventsObserver adapts a Metrics into a watchevents.Observer +// (BUG-2727), so internal/watchevents reports its operational events +// without importing Prometheus. +// +// This is an ADAPTER rather than a bus wrapper — the shape +// NewInstrumentedBus uses for events.EventBus — because every condition it +// reports is detected on the RECEIVE path, inside the bus, and is invisible +// from outside the Bus interface. A wrapper can count publishes and +// subscribers; it cannot see a notification that never arrived. +type WatchEventsObserver struct { + m *Metrics +} + +// NewWatchEventsObserver returns an observer writing into m. +func NewWatchEventsObserver(m *Metrics) *WatchEventsObserver { + return &WatchEventsObserver{m: m} +} + +// Compile-time proof this satisfies the seam it exists for. Without it, a +// later signature change in the interface would surface as a wiring error +// in cmd/pad rather than here. +var _ watchevents.Observer = (*WatchEventsObserver)(nil) + +func (o *WatchEventsObserver) NotificationDropped(reason string) { + o.m.WatchNotificationsDroppedTotal.WithLabelValues(reason).Inc() +} + +// SequenceGap increments both counters: the gap event, and the number of +// notifications it spans. A non-positive `missing` still counts as an +// event — the bus only calls this on a forward jump, so a value that is +// not positive means the caller's arithmetic changed, and swallowing it +// would hide that regression rather than the gap. +func (o *WatchEventsObserver) SequenceGap(missing int64) { + o.m.WatchSequenceGapsTotal.Inc() + if missing > 0 { + o.m.WatchNotificationsMissedTotal.Add(float64(missing)) + } +} + +func (o *WatchEventsObserver) SequenceReset(reason string) { + o.m.WatchSequenceResetsTotal.WithLabelValues(reason).Inc() +} + +func (o *WatchEventsObserver) ReceiveLoopExited() { + o.m.WatchReceiveLoopExitsTotal.Inc() +} diff --git a/internal/server/handlers_workspaces.go b/internal/server/handlers_workspaces.go index 42cd260f7..e3156fa54 100644 --- a/internal/server/handlers_workspaces.go +++ b/internal/server/handlers_workspaces.go @@ -130,6 +130,39 @@ func (s *Server) handleHealthReady(w http.ResponseWriter, r *http.Request) { "driver": string(s.store.D().Driver()), } + // Redis reachability, INFORMATIONAL — it never changes the 200/503 + // (BUG-2727). Readiness stays database-only because every write path, + // the REST API and the web UI work with Redis down; gating on it would + // pull healthy replicas out of the load balancer over a degraded + // feature. What Redis being down actually costs is named on the + // `degrades` field so a reader does not have to know the architecture + // to interpret the flag. + // + // Absent when no prober is wired, which means no Redis is configured. + // A `false` there would say "Redis is down" about a deployment that + // has none. + if s.redisHealth != nil { + status := s.redisHealth.Status() + redisResp := map[string]interface{}{ + "reachable": status.Reachable, + "probed": status.Probed, + } + if status.Error != "" { + redisResp["error"] = status.Error + } + if !status.LastCheck.IsZero() { + redisResp["last_check"] = status.LastCheck.UTC().Format(time.RFC3339) + } + if status.Probed && !status.Reachable { + redisResp["degrades"] = []string{ + "cross-instance activity events", + "watch notifications", + "session presence and session-targeted push", + } + } + resp["redis"] = redisResp + } + writeJSON(w, http.StatusOK, resp) } diff --git a/internal/server/redis_health.go b/internal/server/redis_health.go new file mode 100644 index 000000000..12297c456 --- /dev/null +++ b/internal/server/redis_health.go @@ -0,0 +1,203 @@ +package server + +import ( + "context" + "log/slog" + "sync" + "time" + + "github.com/redis/go-redis/v9" +) + +const ( + // redisProbeInterval is how often the prober pings Redis. Chosen to be + // a few times finer than the presence TTL (90s) so a loss of Redis is + // visible in the health payload and in pad_redis_up well before the + // first user-visible consequence — a session expiring out of the + // picker — rather than at the same moment. + redisProbeInterval = 15 * time.Second + + // redisProbeTimeout bounds one ping ONCE A CONNECTION EXISTS. + // Deliberately short: this is a liveness question, and a probe that + // hangs is itself the answer. + // + // It does NOT bound the whole probe — see RedisHealth's doc comment. + // Establishing a connection is governed by the client's DialTimeout + // (go-redis v9 default: 5s), which this context cannot shorten, so the + // real worst case for one probe is that dial rather than this value. + // redisProbeInterval is comfortably longer than the dial bound, so a + // probe against a black-holed server still finishes before the next + // tick is due. + redisProbeTimeout = 2 * time.Second +) + +// RedisHealth periodically probes Redis and caches the result (BUG-2727). +// +// WHY A PROBER AND NOT A SCRAPE-TIME COLLECTOR. Both /health/ready and +// /metrics need to answer "is Redis reachable", and both are called by +// automated systems on their own schedules. Dialling from inside either +// one makes Pad's Redis load a function of how many monitors are pointed +// at it, and makes the answer depend on WHO ASKED — a scrape and a +// readiness check a millisecond apart could disagree. One prober on a +// fixed interval, two readers of its cached result, and the reported value +// is the same for everyone. +// +// WHAT IT DOES NOT DO. It does not gate readiness. /health/ready stays +// database-only, deliberately: the REST API, the web UI and every write +// path work with Redis down, so failing readiness on a Redis blip would +// pull healthy replicas out of the load balancer and convert a degraded +// feature into an outage. This type exists to make the degradation +// VISIBLE, not to act on it. +// +// TIMEOUT NOTE, measured on BUG-2698's day-50 run rather than assumed: +// go-redis does NOT apply a command context to connection ESTABLISHMENT. +// A call with a 150ms context against a server that accepts connections +// and never answers took 5.0s — the client's DialTimeout. So the probe's +// context bounds the PING once a connection exists and nothing else; the +// dial is bounded by the client's own DialTimeout (go-redis v9 default +// 5s, and cmd/pad/cmd_server.go takes the default). That is why the probe +// interval is set well above the dial bound rather than just above the +// probe timeout. +type RedisHealth struct { + client *redis.Client + interval time.Duration + timeout time.Duration + + // onProbe, when non-nil, is called after every probe with its result. + // It is how the Prometheus gauge is written without this type + // importing the metrics package. + onProbe func(ok bool) + + mu sync.RWMutex + probed bool + ok bool + lastErr string + lastCheck time.Time + + cancel context.CancelFunc + wg sync.WaitGroup +} + +// RedisHealthStatus is the cached probe result, as reported by +// /health/ready. +type RedisHealthStatus struct { + // Reachable is the last probe's verdict. Meaningless until Probed. + Reachable bool `json:"reachable"` + // Probed is false in the window between the server accepting traffic + // and the first probe completing. Reported rather than defaulted: + // "not measured yet" and "measured, and it is down" are different + // facts, and only one of them is worth waking someone for. + Probed bool `json:"probed"` + // Error is the last probe's failure text, empty when Reachable. + Error string `json:"error,omitempty"` + // LastCheck is when the last probe completed, zero before the first. + LastCheck time.Time `json:"last_check,omitempty"` +} + +// NewRedisHealth builds a prober for client. onProbe may be nil. +func NewRedisHealth(client *redis.Client, onProbe func(ok bool)) *RedisHealth { + return &RedisHealth{ + client: client, + interval: redisProbeInterval, + timeout: redisProbeTimeout, + onProbe: onProbe, + } +} + +// Start begins probing in the background. It runs one probe SYNCHRONOUSLY +// before returning, so a server that has finished starting has a real +// answer rather than a Probed=false window that only closes an interval +// later — the boot case is exactly when an operator is watching. +func (h *RedisHealth) Start() { + if h == nil || h.client == nil { + return + } + ctx, cancel := context.WithCancel(context.Background()) + h.cancel = cancel + + h.probe(ctx) + + h.wg.Add(1) + go func() { + defer h.wg.Done() + ticker := time.NewTicker(h.interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + h.probe(ctx) + } + } + }() +} + +// Stop ends probing and waits for the goroutine to exit. Idempotent. +func (h *RedisHealth) Stop() { + if h == nil || h.cancel == nil { + return + } + h.cancel() + h.wg.Wait() +} + +func (h *RedisHealth) probe(ctx context.Context) { + probeCtx, cancel := context.WithTimeout(ctx, h.timeout) + err := h.client.Ping(probeCtx).Err() + cancel() + + // A cancelled parent means we are shutting down, not that Redis is + // gone. Recording it as a failure would flip pad_redis_up to 0 on + // every clean shutdown and teach whoever alerts on it to ignore the + // signal. + if ctx.Err() != nil { + return + } + + ok := err == nil + + h.mu.Lock() + wasProbed, was := h.probed, h.ok + h.probed = true + h.ok = ok + h.lastCheck = time.Now() + if ok { + h.lastErr = "" + } else { + h.lastErr = err.Error() + } + h.mu.Unlock() + + // Log only on TRANSITIONS (and on the first probe), not every tick: a + // Redis outage lasting an hour would otherwise write 240 identical + // lines, which is the same log-flood shape the presence renewal + // warning had to be throttled for. + switch { + case !wasProbed && !ok: + slog.Error("redis: unreachable at startup; watch notifications, cross-instance events and session presence are degraded", "error", err) + case wasProbed && was && !ok: + slog.Error("redis: became unreachable; watch notifications, cross-instance events and session presence are degraded", "error", err) + case wasProbed && !was && ok: + slog.Info("redis: reachable again") + } + + if h.onProbe != nil { + h.onProbe(ok) + } +} + +// Status returns the cached probe result. +func (h *RedisHealth) Status() RedisHealthStatus { + if h == nil { + return RedisHealthStatus{} + } + h.mu.RLock() + defer h.mu.RUnlock() + return RedisHealthStatus{ + Reachable: h.ok, + Probed: h.probed, + Error: h.lastErr, + LastCheck: h.lastCheck, + } +} diff --git a/internal/server/server.go b/internal/server/server.go index e06d42bbd..a1a3ddf38 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -48,6 +48,7 @@ type Server struct { events events.EventBus // real-time event bus (optional) watchEvents watchevents.Bus // watch/nudge notification bus (optional, TASK-2533) sessionPresence SessionPresence // live event-stream connections per user (optional, PLAN-2558 S1) + redisHealth *RedisHealth // cached Redis reachability, reported by /health/ready and pad_redis_up (optional, BUG-2727) collab *collab.RoomManager // Yjs collab room manager (PLAN-1248); optional webhooks *webhooks.Dispatcher // webhook dispatcher (optional) email *email.Sender // transactional email sender (optional) @@ -795,6 +796,18 @@ func (s *Server) SetSessionPresence(p SessionPresence) { s.sessionPresence = p } +// SetRedisHealth attaches the Redis reachability prober (BUG-2727). Nil +// on a deployment with no Redis, in which case /health/ready reports no +// redis block at all — the honest shape, since "no Redis configured" and +// "Redis configured and down" are different states and a `false` would +// merge them. +// +// It does NOT make readiness depend on Redis; see RedisHealth's doc +// comment for why that is deliberate. +func (s *Server) SetRedisHealth(h *RedisHealth) { + s.redisHealth = h +} + // SetCollabRoomManager attaches a Yjs collab RoomManager (PLAN-1248). // When set, the /api/v1/collab/{itemID} WebSocket endpoint hands new // connections to the manager for op-log replay + fan-out. When nil, diff --git a/internal/server/session_presence_observer.go b/internal/server/session_presence_observer.go new file mode 100644 index 000000000..34ea07f45 --- /dev/null +++ b/internal/server/session_presence_observer.go @@ -0,0 +1,91 @@ +package server + +import ( + "sync" + + "github.com/PerpetualSoftware/pad/internal/metrics" +) + +// PresenceObserver receives operational events from the Redis-backed +// session-presence registry (BUG-2727). +// +// The registry is deliberately FAIL-SOFT everywhere: a failed write leaves +// a live session unlisted rather than tearing down the SSE connection it +// belongs to, and a failed read answers the sessions endpoint with a 503 +// rather than an empty list. That posture is right, and it is exactly why +// the failures need a counter — a fail-soft subsystem reports nothing to +// the user beyond a push that quietly reaches fewer sessions than it +// should, so a log line is the only trace and log lines are not alertable +// without someone already looking. +// +// Implementations must be safe for concurrent use and must not block. +type PresenceObserver interface { + // PresenceOpFailed reports one failed Redis operation. op is a bounded + // label from the PresenceOp* constants, never free text — it becomes a + // metric label. + PresenceOpFailed(op string) +} + +// Presence operation labels. Bounded by construction so they are safe as +// metric label values. +const ( + // PresenceOpRegister — the initial write for a newly connected + // session. It leaves the session out of the picker until the first + // successful renewal. + PresenceOpRegister = "register" + // PresenceOpRenew — the keepalive write. Sustained failure lets the + // entry expire on its TTL while the connection is still live. + PresenceOpRenew = "renew" + // PresenceOpDeregister — the explicit removal on disconnect. Failure + // leaves a dead session listed until its TTL expires, so the picker + // over-reports. + PresenceOpDeregister = "deregister" + // PresenceOpList — reading a user's sessions. Failure surfaces to the + // caller as a 503 rather than as a short list. + PresenceOpList = "list" + // PresenceOpPrune — removing index members whose session key is gone. + // Failure is the most benign: the index carries a stale member that + // the next read skips and the next prune retries. + PresenceOpPrune = "prune" +) + +// presenceObservable is the nil-safe holder RedisSessionPresence embeds. +type presenceObservable struct { + obsMu sync.RWMutex + obs PresenceObserver +} + +// SetObserver attaches a PresenceObserver; nil detaches. Safe to call +// after the registry is running. +func (o *presenceObservable) SetObserver(obs PresenceObserver) { + o.obsMu.Lock() + o.obs = obs + o.obsMu.Unlock() +} + +func (o *presenceObservable) reportOpFailed(op string) { + o.obsMu.RLock() + obs := o.obs + o.obsMu.RUnlock() + if obs != nil { + obs.PresenceOpFailed(op) + } +} + +// metricsPresenceObserver adapts the Prometheus metrics into a +// PresenceObserver. It lives here rather than in internal/metrics because +// internal/server already imports internal/metrics; the reverse would be +// an import cycle. +type metricsPresenceObserver struct { + m *metrics.Metrics +} + +func (o metricsPresenceObserver) PresenceOpFailed(op string) { + o.m.SessionPresenceFailuresTotal.WithLabelValues(op).Inc() +} + +// NewMetricsPresenceObserver returns a PresenceObserver that writes into +// m's Prometheus counters. Exported because the wiring lives in cmd/pad. +func NewMetricsPresenceObserver(m *metrics.Metrics) PresenceObserver { + return metricsPresenceObserver{m: m} +} diff --git a/internal/server/session_presence_redis.go b/internal/server/session_presence_redis.go index 6dee89620..868478182 100644 --- a/internal/server/session_presence_redis.go +++ b/internal/server/session_presence_redis.go @@ -129,6 +129,10 @@ import ( // busy. Consumers must keep treating this list as "connected as far as the // server can tell", never as a delivery guarantee. type RedisSessionPresence struct { + // presenceObservable carries the optional operational-event seam + // (BUG-2727). Embedded so SetObserver is part of the type's surface. + presenceObservable + client *redis.Client // sessionKeyTTL and renewInterval are fields rather than constants so @@ -447,6 +451,7 @@ func (p *RedisSessionPresence) Add(userID string, ident SessionIdentity) string // registration that failed on a blip. slog.Warn("session presence: failed to register session; it will be missing from the picker until the next renewal", "error", err, "user_id", userID) + p.reportOpFailed(PresenceOpRegister) } go p.renewLoop(ctx, rn, userID, id, string(payload)) @@ -512,6 +517,14 @@ func (p *RedisSessionPresence) renewLoop(ctx context.Context, rn *renewal, userI } if err := p.write(ctx, userID, sessionID, payload); err != nil && ctx.Err() == nil { p.warnRenewFailure(err, userID, sessionID) + // Counted on EVERY failure, while the log line above is + // rate-limited (that throttle exists because a Redis + // outage otherwise logs once per session per renewal — + // roughly 33 lines a second per replica at a thousand + // sessions). A counter has no such problem and wants the + // true rate: throttling it too would make the metric + // under-report exactly during the incident it exists for. + p.reportOpFailed(PresenceOpRenew) } } } @@ -588,6 +601,7 @@ func (p *RedisSessionPresence) Remove(userID string, sessionID string) { if _, err := pipe.Exec(ctx); err != nil { slog.Warn("session presence: failed to deregister session; it will expire on its TTL", "error", err, "user_id", userID) + p.reportOpFailed(PresenceOpDeregister) } } @@ -617,6 +631,7 @@ func (p *RedisSessionPresence) ListForUser(userID string) ([]LiveSession, error) // anything. Collapsing the two here is the same defect this type // was written to remove, one layer down. slog.Warn("session presence: failed to read session index", "error", err, "user_id", userID) + p.reportOpFailed(PresenceOpList) return nil, fmt.Errorf("session presence: read index: %w", err) } if len(ids) == 0 { @@ -638,6 +653,7 @@ func (p *RedisSessionPresence) ListForUser(userID string) ([]LiveSession, error) values, err := p.client.MGet(ctx, keys...).Result() if err != nil { slog.Warn("session presence: failed to read session entries", "error", err, "user_id", userID) + p.reportOpFailed(PresenceOpList) return nil, fmt.Errorf("session presence: read entries: %w", err) } @@ -701,6 +717,7 @@ func (p *RedisSessionPresence) ListForUser(userID string) ([]LiveSession, error) ).Err(); err != nil && !errors.Is(err, redis.Nil) { slog.Debug("session presence: failed to prune expired index members", "error", err, "user_id", userID) + p.reportOpFailed(PresenceOpPrune) } } diff --git a/internal/watchevents/observer.go b/internal/watchevents/observer.go new file mode 100644 index 000000000..09f69c2f6 --- /dev/null +++ b/internal/watchevents/observer.go @@ -0,0 +1,121 @@ +package watchevents + +import "sync" + +// Observer receives OPERATIONAL events from a Bus — the conditions an +// operator would want to alert on, as opposed to the notifications the bus +// exists to deliver. It is the seam internal/metrics implements so this +// package can stay free of a Prometheus dependency, mirroring the reason +// events.EventBus is instrumented by a wrapper rather than by importing +// metrics. +// +// A wrapper does not work HERE, though, and that asymmetry is the whole +// reason this interface exists: every condition below is detected INSIDE +// the bus, on the receive path, and is invisible at the Bus interface. A +// caller sees Publish succeed and Subscribe return a channel whether or +// not this instance is quietly missing notifications. +// +// WHAT THIS DOES AND DOES NOT SEE (BUG-2727). These callbacks report what +// PAD detects. They do NOT hook go-redis, which has its own silent drop: +// it buffers 100 messages per subscription and discards further ones after +// a 60s send timeout, logging only through its own internal logger. That +// drop is not reported here directly — it is reported by its CONSEQUENCE, +// as a SequenceGap, because a discarded message leaves a hole in the id +// sequence exactly like a message lost across a reconnect does. So +// SequenceGap counts "this instance missed something", not "Redis was at +// fault"; do not read a gap as evidence of any particular cause. +// +// Implementations must be safe for concurrent use and must not block — +// they are called while the bus holds its mutex on the receive path. +type Observer interface { + // NotificationDropped reports a notification this instance received but + // could not deliver to one of its own subscribers. reason is a bounded + // label ("slow_subscriber"), never free text — it becomes a metric + // label. + NotificationDropped(reason string) + + // SequenceGap reports that the received id sequence skipped forward: + // this instance missed `missing` notifications. Resumes across the gap + // are answered with sync_required (see fanOutLocally); subscribers + // holding a stream open across it are told nothing, which is BUG-2730 + // and not this seam's job to fix. + SequenceGap(missing int64) + + // SequenceReset reports that the id space itself changed under this + // instance and the replay buffer was dropped. reason is bounded: + // "epoch_change" (the shared epoch key changed) or "counter_backwards" + // (an id arrived at or below the high-water mark). + SequenceReset(reason string) + + // ReceiveLoopExited reports that the single consumer of the shared + // Redis channel has stopped. After this the instance publishes fine and + // receives nothing — including its own publishes, which are delivered + // through Redis like everyone else's. + // + // Reachability, verified against go-redis v9.22.0 rather than assumed: + // PubSub.Channel's message channel is closed ONLY on pool.ErrClosed, + // i.e. the client or the PubSub was closed. Every other receive error + // is retried indefinitely, and a health-check goroutine pings every 3s + // and reconnects (re-subscribing) on failure. So in practice this fires + // at shutdown, and a firing OUTSIDE shutdown means the client was + // closed underneath the bus — which is why it is worth a counter and an + // ERROR log rather than a silent return. + ReceiveLoopExited() +} + +// observable is the shared, nil-safe Observer holder both bus +// implementations embed. Reporting through it before SetObserver is +// called — which is every bus in every test that does not opt in, and +// every single-process deployment — is a no-op. +type observable struct { + obsMu sync.RWMutex + obs Observer +} + +// SetObserver attaches an Observer. Passing nil detaches. Safe to call +// after the bus is running; cmd/pad/cmd_server.go calls it at wiring time. +func (o *observable) SetObserver(obs Observer) { + o.obsMu.Lock() + o.obs = obs + o.obsMu.Unlock() +} + +func (o *observable) observer() Observer { + o.obsMu.RLock() + obs := o.obs + o.obsMu.RUnlock() + return obs +} + +func (o *observable) reportDropped(reason string) { + if obs := o.observer(); obs != nil { + obs.NotificationDropped(reason) + } +} + +func (o *observable) reportGap(missing int64) { + if obs := o.observer(); obs != nil { + obs.SequenceGap(missing) + } +} + +func (o *observable) reportReset(reason string) { + if obs := o.observer(); obs != nil { + obs.SequenceReset(reason) + } +} + +func (o *observable) reportReceiveLoopExited() { + if obs := o.observer(); obs != nil { + obs.ReceiveLoopExited() + } +} + +// Drop reasons and reset reasons. Bounded by construction so they can be +// metric labels without a cardinality risk. +const ( + DropReasonSlowSubscriber = "slow_subscriber" + + ResetReasonEpochChange = "epoch_change" + ResetReasonCounterBackward = "counter_backwards" +) diff --git a/internal/watchevents/observer_test.go b/internal/watchevents/observer_test.go new file mode 100644 index 000000000..9d5292731 --- /dev/null +++ b/internal/watchevents/observer_test.go @@ -0,0 +1,267 @@ +package watchevents + +import ( + "sync" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" +) + +// recordingObserver is the instrument under which the tests below run. It +// counts, rather than merely noting, so a test can distinguish "reported +// once" from "reported on every subscriber in the loop" — a real +// difference, since the drop site lives inside a per-subscriber loop. +type recordingObserver struct { + mu sync.Mutex + + dropped map[string]int + gaps int + missed int64 + resets map[string]int + loopExits int + totalEvents int +} + +func newRecordingObserver() *recordingObserver { + return &recordingObserver{ + dropped: map[string]int{}, + resets: map[string]int{}, + } +} + +func (o *recordingObserver) NotificationDropped(reason string) { + o.mu.Lock() + defer o.mu.Unlock() + o.dropped[reason]++ + o.totalEvents++ +} + +func (o *recordingObserver) SequenceGap(missing int64) { + o.mu.Lock() + defer o.mu.Unlock() + o.gaps++ + o.missed += missing + o.totalEvents++ +} + +func (o *recordingObserver) SequenceReset(reason string) { + o.mu.Lock() + defer o.mu.Unlock() + o.resets[reason]++ + o.totalEvents++ +} + +func (o *recordingObserver) ReceiveLoopExited() { + o.mu.Lock() + defer o.mu.Unlock() + o.loopExits++ + o.totalEvents++ +} + +func (o *recordingObserver) snapshot() recordingObserver { + o.mu.Lock() + defer o.mu.Unlock() + cp := recordingObserver{ + dropped: map[string]int{}, + resets: map[string]int{}, + gaps: o.gaps, + missed: o.missed, + loopExits: o.loopExits, + totalEvents: o.totalEvents, + } + for k, v := range o.dropped { + cp.dropped[k] = v + } + for k, v := range o.resets { + cp.resets[k] = v + } + return cp +} + +// TestMemoryBusReportsSlowSubscriberDrop asserts the DROP increments the +// observer — not merely that an observer can be attached. +// +// It asserts its own premise first: a subscriber with room receives the +// notification and produces NO drop report. Without that leg the test +// would pass against a bus that reported a drop on every publish, which +// is a differently-broken instrument that looks identical from the +// assertion that only checks the full case. +func TestMemoryBusReportsSlowSubscriberDrop(t *testing.T) { + t.Parallel() + + b := New() + defer b.Close() + obs := newRecordingObserver() + b.SetObserver(obs) + + ch := b.Subscribe() + + // PREMISE: a healthy subscriber is delivered to and reports nothing. + if err := b.Publish(Notification{Kind: KindPush, ItemRef: "TASK-1"}); err != nil { + t.Fatalf("publish: %v", err) + } + select { + case <-ch: + case <-time.After(time.Second): + t.Fatal("premise failed: a subscriber with room did not receive the notification") + } + if got := obs.snapshot(); got.totalEvents != 0 { + t.Fatalf("premise failed: healthy delivery reported %d observer events, want 0 (%+v)", got.totalEvents, got) + } + + // Fill the subscriber's 64-deep buffer without draining it. + for i := 0; i < 64; i++ { + if err := b.Publish(Notification{Kind: KindPush, ItemRef: "TASK-2"}); err != nil { + t.Fatalf("publish %d: %v", i, err) + } + } + if got := obs.snapshot(); got.totalEvents != 0 { + t.Fatalf("premise failed: filling the buffer to capacity reported %d events, want 0 (%+v)", got.totalEvents, got) + } + + // The 65th has nowhere to go. + if err := b.Publish(Notification{Kind: KindPush, ItemRef: "TASK-3"}); err != nil { + t.Fatalf("overflow publish: %v", err) + } + + got := obs.snapshot() + if got.dropped[DropReasonSlowSubscriber] != 1 { + t.Fatalf("dropped[%s] = %d, want 1 (%+v)", DropReasonSlowSubscriber, got.dropped[DropReasonSlowSubscriber], got) + } + if got.totalEvents != 1 { + t.Fatalf("total observer events = %d, want exactly 1 — the drop and nothing else (%+v)", got.totalEvents, got) + } +} + +// TestRedisBusReportsSequenceGap drives fanOutLocally directly, which is +// where the detection lives, and pins BOTH numbers: that a gap was +// reported once, and how many notifications it spanned. One counter +// cannot distinguish one gap of 500 from 500 gaps of one, which is why +// the observer reports both and why this asserts both. +func TestRedisBusReportsSequenceGap(t *testing.T) { + t.Parallel() + + b, _ := newMiniredisBus(t, 64) + obs := newRecordingObserver() + b.SetObserver(obs) + + // PREMISE: contiguous ids report nothing. Without this leg a bus that + // reported a gap on every notification would pass the assertion below. + b.fanOutLocally(Notification{ID: 1, Kind: KindPush}) + b.fanOutLocally(Notification{ID: 2, Kind: KindPush}) + if got := obs.snapshot(); got.totalEvents != 0 { + t.Fatalf("premise failed: contiguous ids 1,2 reported %d observer events, want 0 (%+v)", got.totalEvents, got) + } + + // 3 and 4 are missing. + b.fanOutLocally(Notification{ID: 5, Kind: KindPush}) + + got := obs.snapshot() + if got.gaps != 1 { + t.Fatalf("gaps = %d, want 1 (%+v)", got.gaps, got) + } + if got.missed != 2 { + t.Fatalf("missed = %d, want 2 — ids 3 and 4 (%+v)", got.missed, got) + } +} + +// TestRedisBusReportsSequenceResets covers both reset reasons, which are +// detected in different functions (fanOutFromRedis for the epoch, +// fanOutLocally for the counter) and would be easy to wire one of and +// call done. +func TestRedisBusReportsSequenceResets(t *testing.T) { + t.Parallel() + + t.Run("epoch change", func(t *testing.T) { + b, _ := newMiniredisBus(t, 64) + obs := newRecordingObserver() + b.SetObserver(obs) + + b.fanOutFromRedis("epoch-a", Notification{ID: 1, Kind: KindPush}) + if got := obs.snapshot(); got.totalEvents != 0 { + t.Fatalf("premise failed: the first epoch seen reported %d events, want 0 (%+v)", got.totalEvents, got) + } + + b.fanOutFromRedis("epoch-b", Notification{ID: 1, Kind: KindPush}) + + got := obs.snapshot() + if got.resets[ResetReasonEpochChange] != 1 { + t.Fatalf("resets[%s] = %d, want 1 (%+v)", ResetReasonEpochChange, got.resets[ResetReasonEpochChange], got) + } + }) + + t.Run("counter backwards", func(t *testing.T) { + b, _ := newMiniredisBus(t, 64) + obs := newRecordingObserver() + b.SetObserver(obs) + + b.fanOutLocally(Notification{ID: 100, Kind: KindPush}) + if got := obs.snapshot(); got.totalEvents != 0 { + t.Fatalf("premise failed: a cold start reported %d events, want 0 (%+v)", got.totalEvents, got) + } + + // The shared counter restarted: an id at or below the high-water + // mark. + b.fanOutLocally(Notification{ID: 1, Kind: KindPush}) + + got := obs.snapshot() + if got.resets[ResetReasonCounterBackward] != 1 { + t.Fatalf("resets[%s] = %d, want 1 (%+v)", ResetReasonCounterBackward, got.resets[ResetReasonCounterBackward], got) + } + if got.gaps != 0 { + t.Fatalf("gaps = %d, want 0 — a backwards counter is a reset, not a gap (%+v)", got.gaps, got) + } + }) +} + +// TestRedisBusReportsReceiveLoopExit closes the client underneath a +// running bus, which is the ONE condition go-redis actually closes a +// subscription's message channel for (pool.ErrClosed — every other +// receive error is retried indefinitely and a health-check goroutine +// reconnects). Driving the real condition rather than calling the +// reporter directly is the difference between testing the wiring and +// testing the constant. +func TestRedisBusReportsReceiveLoopExit(t *testing.T) { + t.Parallel() + + mr := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + + b := NewRedisBusWithReplaySize(client, 64) + defer b.Close() + obs := newRecordingObserver() + b.SetObserver(obs) + + // PREMISE: the bus is genuinely receiving before we break it — + // otherwise a bus that never started would pass by reporting an exit + // for the wrong reason. + ch := b.Subscribe() + if err := b.Publish(Notification{Kind: KindPush, ItemRef: "TASK-1"}); err != nil { + t.Fatalf("premise publish: %v", err) + } + select { + case <-ch: + case <-time.After(3 * time.Second): + t.Fatal("premise failed: the bus never delivered a round-tripped notification") + } + if got := obs.snapshot(); got.loopExits != 0 { + t.Fatalf("premise failed: %d loop exits before closing the client (%+v)", got.loopExits, got) + } + + // Close the CLIENT, not the bus: b.ctx stays live, so the receive + // loop takes the closed-channel branch rather than the ctx branch. + if err := client.Close(); err != nil { + t.Fatalf("close client: %v", err) + } + + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if obs.snapshot().loopExits == 1 { + return + } + time.Sleep(2 * time.Millisecond) + } + t.Fatalf("receive loop exit was never reported (%+v)", obs.snapshot()) +} diff --git a/internal/watchevents/redis_bus.go b/internal/watchevents/redis_bus.go index c86b60548..0ac400855 100644 --- a/internal/watchevents/redis_bus.go +++ b/internal/watchevents/redis_bus.go @@ -172,6 +172,11 @@ const ( // is bounded: DefaultReplayBufferSize entries, after which a resume answers // "gap too large" and the consumer resyncs. type RedisBus struct { + // observable carries the optional operational-event seam (BUG-2727). + // Embedded first so SetObserver is part of the type's own surface + // rather than something a caller reaches through a field. + observable + client *redis.Client // mu guards subscribers AND replay together — see the type comment (3) @@ -695,6 +700,22 @@ func (b *RedisBus) receiveMessages() { return case msg, ok := <-ch: if !ok { + // The subscription's message channel closed. go-redis + // (v9.22.0) closes it ONLY on pool.ErrClosed — the client + // or the PubSub was closed — and retries every other + // receive error indefinitely while a health-check + // goroutine reconnects on ping failure. So the ordinary + // cause is our own shutdown, which cancels b.ctx and is + // caught by the case above; arriving HERE instead means + // the client went away underneath a bus that is still + // running, and from this point the instance publishes + // fine and receives nothing at all — including its own + // publishes, which come back through Redis like everyone + // else's. Silence was the original behaviour and made + // that state indistinguishable from a quiet workspace. + slog.Error("watchevents: Redis subscription closed; this instance will receive no further notifications " + + "(publishes still succeed, so nothing else will report this)") + b.reportReceiveLoopExited() return } epoch, n, err := decodePayload(msg.Payload) @@ -768,6 +789,7 @@ func (b *RedisBus) fanOutFromRedis(epoch string, n Notification) { b.lastAppendedID = 0 b.knownFrom = 0 b.epochJustChanged = true + b.reportReset(ResetReasonEpochChange) } b.mu.Unlock() @@ -837,11 +859,14 @@ func (b *RedisBus) fanOutLocally(n Notification) { "previous", b.lastAppendedID, "got", n.ID) b.replay = newReplayBuffer(b.replaySize) b.knownFrom = n.ID + b.reportReset(ResetReasonCounterBackward) case n.ID != b.lastAppendedID+1: slog.Warn("watchevents: gap in the received notification sequence; resumes across it will report sync_required", - "expected", b.lastAppendedID+1, "got", n.ID) + "expected", b.lastAppendedID+1, "got", n.ID, + "missed", n.ID-b.lastAppendedID-1) b.knownFrom = n.ID + b.reportGap(n.ID - b.lastAppendedID - 1) } b.lastAppendedID = n.ID @@ -853,6 +878,7 @@ func (b *RedisBus) fanOutLocally(n Notification) { default: slog.Warn("watchevents: dropping notification for slow subscriber", "kind", n.Kind, "item_ref", n.ItemRef) + b.reportDropped(DropReasonSlowSubscriber) } } } diff --git a/internal/watchevents/watchevents.go b/internal/watchevents/watchevents.go index d74257aca..8a348bf27 100644 --- a/internal/watchevents/watchevents.go +++ b/internal/watchevents/watchevents.go @@ -328,6 +328,13 @@ func (rb *replayBuffer) since(sinceID int64) []Notification { // (codex round 2 finding 3 — see Publish's doc comment for why an // earlier "send after releasing the lock" version could panic). type MemoryBus struct { + // observable carries the optional operational-event seam (BUG-2727). + // A single-process deployment never wires one, so every report below + // is a nil check; it is here so the drop condition is reported by the + // SAME name on both implementations rather than existing only on the + // one that happens to have Redis behind it. + observable + mu sync.Mutex subscribers map[chan Notification]struct{} seq int64 @@ -412,6 +419,7 @@ func (b *MemoryBus) Publish(n Notification) error { // a delivery it cannot confirm either way had definitely // failed — exactly the false precision BUG-2699 is about. slog.Warn("watchevents: dropping notification for slow subscriber", "kind", n.Kind, "item_ref", n.ItemRef) + b.reportDropped(DropReasonSlowSubscriber) } } return nil