From dd1f6cde0b3e565965576275f673473c41140d1c Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 26 Feb 2026 20:56:55 -0500 Subject: [PATCH] fix: replication send failure counter and go subscribe error propagation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit **h2 — silent replication sends**: the `broadcast_replication` helper in aof.rs previously discarded send errors with `let _ = tx.send(...)`. now counts failed sends as `ember_replication_send_failures_total` so operators can alert on unexpected replica disconnects in a replicated deployment. adds `metrics` as a workspace dependency in ember-core. also documents the expected-silent try_send pattern in blocking.rs. **h3 — go subscribe error propagation**: the subscribe goroutine previously returned silently on any recv error, leaving callers unable to distinguish normal EOF from network or server errors. introduces a `Subscription` type with a `C` event channel and an `Err()` method, and changes `Subscribe` to return `(*Subscription, error)`. the goroutine now sends the recv error to an internal buffered channel before exiting, which `Err()` exposes after C is drained. --- Cargo.lock | 1 + clients/ember-go/ember.go | 38 +++++++++++++++++++++---- crates/ember-core/Cargo.toml | 1 + crates/ember-core/src/shard/aof.rs | 18 ++++++++---- crates/ember-core/src/shard/blocking.rs | 4 ++- 5 files changed, 50 insertions(+), 12 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cc2e7f63..cdcd340f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1003,6 +1003,7 @@ dependencies = [ "ember-cluster", "ember-persistence", "ember-protocol", + "metrics", "ordered-float", "parking_lot", "prost-reflect", diff --git a/clients/ember-go/ember.go b/clients/ember-go/ember.go index a1018fe7..b040157b 100644 --- a/clients/ember-go/ember.go +++ b/clients/ember-go/ember.go @@ -568,10 +568,30 @@ type SubscribeEvent struct { Pattern string // only set for pmessage } +// Subscription holds the channels returned by [Client.Subscribe]. +// Events are read from C; any stream error that ends the subscription +// is retrievable via Err() after C is closed. +type Subscription struct { + // C receives events until the stream ends or the context is cancelled. + C <-chan SubscribeEvent + errc <-chan error +} + +// Err returns the first error that caused the subscription stream to close, +// or nil if the stream ended cleanly (context cancellation or EOF). +// Call Err only after C has been drained and closed. +func (s *Subscription) Err() error { + if err, ok := <-s.errc; ok { + return err + } + return nil +} + // Subscribe opens a server-streaming subscription for the given channels -// and/or patterns. Returns a channel that yields events until the context -// is cancelled or the stream ends. -func (c *Client) Subscribe(ctx context.Context, channels []string, patterns []string) (<-chan SubscribeEvent, error) { +// and/or patterns. Returns a Subscription whose C field yields events +// until the context is cancelled or the stream ends. Use sub.Err() after +// the channel closes to check whether the stream ended due to an error. +func (c *Client) Subscribe(ctx context.Context, channels []string, patterns []string) (*Subscription, error) { stream, err := c.rpc.Subscribe(c.ctx(ctx), &pb.SubscribeRequest{ Channels: channels, Patterns: patterns, @@ -581,11 +601,17 @@ func (c *Client) Subscribe(ctx context.Context, channels []string, patterns []st } ch := make(chan SubscribeEvent, 64) + errc := make(chan error, 1) go func() { defer close(ch) + defer close(errc) for { - evt, err := stream.Recv() - if err != nil { + evt, recvErr := stream.Recv() + if recvErr != nil { + if ctx.Err() == nil { + // stream closed unexpectedly — propagate the error + errc <- recvErr + } return } se := SubscribeEvent{ @@ -604,7 +630,7 @@ func (c *Client) Subscribe(ctx context.Context, channels []string, patterns []st } }() - return ch, nil + return &Subscription{C: ch, errc: errc}, nil } // PubSubChannels returns active channel names, optionally filtered by pattern. diff --git a/crates/ember-core/Cargo.toml b/crates/ember-core/Cargo.toml index e7fd447f..6eaa6f82 100644 --- a/crates/ember-core/Cargo.toml +++ b/crates/ember-core/Cargo.toml @@ -23,6 +23,7 @@ ember-protocol = { workspace = true } ember-persistence = { workspace = true } thiserror = { workspace = true } bytes = { workspace = true } +metrics = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } rand = { workspace = true } diff --git a/crates/ember-core/src/shard/aof.rs b/crates/ember-core/src/shard/aof.rs index 937219ab..8b218251 100644 --- a/crates/ember-core/src/shard/aof.rs +++ b/crates/ember-core/src/shard/aof.rs @@ -574,11 +574,19 @@ pub(super) fn broadcast_replication( ) { if let Some(ref tx) = *replication_tx { *replication_offset += 1; - let _ = tx.send(ReplicationEvent { - shard_id, - offset: *replication_offset, - record, - }); + if tx + .send(ReplicationEvent { + shard_id, + offset: *replication_offset, + record, + }) + .is_err() + { + // no replicas are currently connected — normal during startup + // or after a replica disconnects, but tracked so operators + // can alert on unexpected drops in a replicated deployment. + metrics::counter!("ember_replication_send_failures_total").increment(1); + } } } diff --git a/crates/ember-core/src/shard/blocking.rs b/crates/ember-core/src/shard/blocking.rs index 0398039e..a96d7328 100644 --- a/crates/ember-core/src/shard/blocking.rs +++ b/crates/ember-core/src/shard/blocking.rs @@ -25,7 +25,9 @@ pub(super) fn handle_blocking_pop( match result { Ok(Some(data)) => { - // got an element — send to waiter and record the mutation + // got an element — send to waiter and record the mutation. + // try_send can only fail if the client disconnected between + // registering the waiter and the element arriving; safe to ignore. let _ = waiter.try_send((key.to_owned(), data)); reply.send(ShardResponse::Ok);