Go SDK - Stream Observer

Pass a nexus.StreamObserver via Config.Observer to receive structured callbacks for every SSE lifecycle transition. The observer is the canonical way to wire SDK state into ops dashboards, custom telemetry, or load-test tooling - far more reliable than scraping Logger output by substring match. The observer is opt-in. When Config.Observer == nil the SDK uses NoopStreamObserver at zero overhead.

The interface

type StreamObserver interface {
    OnConnected()
    OnDisconnected(err error)
    OnEvent(name string)
    OnReconnectAttempt(attempt int)
    OnFallback(reason string)
    OnQuarantined(reason string, expiresAt time.Time)
}

Embed nexus.NoopStreamObserver to override only the hooks you care about:

type myObserver struct {
    nexus.NoopStreamObserver
    events atomic.Int64
}
func (o *myObserver) OnEvent(string) { o.events.Add(1) }

Hook semantics

Hook	Fires when	Argument
`OnConnected()`	`GET /v1/stream` returns `200 OK` and the SDK is ready to read events	none
`OnDisconnected(err)`	An active stream connection ends - clean cancel, transport error, or server-initiated close	`nil` for clean shutdown, non-nil otherwise
`OnEvent(name)`	A whitelisted SSE event arrives, before the SDK runs the event-triggered sync	event name string (e.g. `"config.updated"`)
`OnReconnectAttempt(n)`	The SDK is about to back off and retry after a transport error	1-based attempt counter (max 3 with current threshold)
`OnFallback(reason)`	The SDK switches to TTL polling or signals an active rate-limit episode it will recover from automatically	`nexus.FallbackReasonMaxErrors` or `nexus.FallbackReasonRateLimited`
`OnQuarantined(reason, expiresAt)`	The backend returned `429` with a quarantine body - the SDK will not retry sync until `expiresAt`	`reason`: a short code describing why the service was quarantined; `expiresAt`: the absolute `time.Time` after which normal sync resumes

What counts as a “whitelisted event”

OnEvent fires only for events on the SDK’s documented event-type list:

resync
config.created / config.updated / config.deleted
flag.created / flag.updated / flag.toggled / flag.deleted
secret.created / secret.updated / secret.deleted

Unknown events (or the WIF auth_expiring control event) do NOT trigger OnEvent.

`OnFallback` reasons

const (
    FallbackReasonMaxErrors   = "max-errors"
    FallbackReasonRateLimited = "rate-limited"
)

Constant	When
`FallbackReasonMaxErrors`	3 consecutive transport errors - backoff exhausted, the SDK gives up and `RunStream` returns
`FallbackReasonRateLimited`	server returned `429 Too Many Requests` (tier connection-limit hit) - see active-recovery note below

`FallbackReasonRateLimited` - active-recovery semantics (v0.4.0+)

FallbackReasonRateLimited is dispatched in both of the following situations:

Terminate case (no Retry-After) - the server returned 429 without a parseable Retry-After header. The SDK closes the stream, RunStream returns, and TTL polling takes over.
Active-recovery case (Retry-After present) - the server returned 429 with a parseable Retry-After header. The SDK fires OnFallback("rate-limited") for visibility, then sleeps the indicated delay (clamped [5 s, 5 min]) and reconnects automatically. RunStream does NOT return, the transport-failure counter is NOT incremented, and the stream resumes without intervention.

Observers that previously treated FallbackReasonRateLimited as a hard “stream is gone” signal should be aware that in v0.4.0+ the stream may recover on its own. A subsequent OnConnected() callback will fire when the reconnect succeeds.

`OnQuarantined` - quarantine semantics (v0.4.0+)

OnQuarantined(reason string, expiresAt time.Time) fires when the backend returns 429 Too Many Requests with a structured quarantine body:

{"error":"quarantined","reason":"abuse-detected","expires_at":"2026-05-28T14:00:00Z"}

SDK behaviour while quarantined:

Background sync (ensureFresh) is skipped entirely - no requests are sent to the backend.
The in-memory cache continues to be served for all reads (GetConfig, GetSecret, GetFlag).
The quarantine state clears automatically when expiresAt passes.

Threading contract

Observer methods are called synchronously from the SDK’s own goroutine. They MUST return promptly. The SDK does not protect against blocking observers.

If your observer does anything substantial (network I/O, mutex contention, disk writes), copy the argument and dispatch the work to your own goroutine before returning:

func (o *slowObserver) OnEvent(name string) {
    select {
    case o.queue <- name:
    default:
        // Drop on overflow - applying backpressure to the SDK goroutine
        // would block sync() and stall reads.
    }
}

Counter pattern

The most common use - counting events for a dashboard or load-test report:

type counterObserver struct {
    nexus.NoopStreamObserver
    events     atomic.Int64
    reconnects atomic.Int64
    fallbacks  atomic.Int64
}
func (c *counterObserver) OnEvent(string)            { c.events.Add(1) }
func (c *counterObserver) OnReconnectAttempt(int)    { c.reconnects.Add(1) }
func (c *counterObserver) OnFallback(string)         { c.fallbacks.Add(1) }

obs := &counterObserver{}
client, _ := nexus.NewClient(ctx, nexus.Config{
    BaseURL:  "...",
    APIKey:   "...",
    Observer: obs,
})

// later ...
fmt.Printf("SSE: events=%d reconnects=%d fallback=%d\n",
    obs.events.Load(), obs.reconnects.Load(), obs.fallbacks.Load())

Active-connection gauge pattern

Maintain a live “is the SDK connected” gauge:

type gaugeObserver struct {
    nexus.NoopStreamObserver
    connected atomic.Bool
}
func (g *gaugeObserver) OnConnected()         { g.connected.Store(true) }
func (g *gaugeObserver) OnDisconnected(error) { g.connected.Store(false) }
func (g *gaugeObserver) OnFallback(string)    { g.connected.Store(false) }

OnFallback is included because after the 3-strike fallback the SDK will not reconnect - the gauge should stay false until you spin up a new client or the next manual RunStream(ctx) call. Note: with FallbackReasonRateLimited (v0.4.0+) the SDK may auto-recover via Retry-After; the subsequent OnConnected will flip the gauge back to true.

Per-event-type counter

type perEventObserver struct {
    nexus.NoopStreamObserver
    mu     sync.Mutex
    counts map[string]int
}
func (p *perEventObserver) OnEvent(name string) {
    p.mu.Lock()
    defer p.mu.Unlock()
    if p.counts == nil {
        p.counts = make(map[string]int)
    }
    p.counts[name]++
}

Independence from `Logger`

The observer is independent of Logger. The two are complementary:

Logger carries human-readable text for operators ("nexus: SSE stream connected").
StreamObserver carries structured metrics for tooling.

A consumer can set both, neither, or just one. Setting only the observer is the right choice when you don’t want any text output from the SDK.

Compatibility note

If you previously relied on substring-matching Logger.Errorf lines to count reconnects/fallbacks, switch to the observer - log line wording is not a stable contract and may change between SDK releases.

Overview

Backend SDKs

Browser SDKs

OpenFeature Providers

Go SDK - Stream Observer

The interface

Hook semantics

What counts as a “whitelisted event”

`OnFallback` reasons

`FallbackReasonRateLimited` - active-recovery semantics (v0.4.0+)

`OnQuarantined` - quarantine semantics (v0.4.0+)

Threading contract

Counter pattern

Active-connection gauge pattern

Per-event-type counter

Independence from `Logger`

Compatibility note

​The interface

​Hook semantics

​What counts as a “whitelisted event”

​OnFallback reasons

​FallbackReasonRateLimited - active-recovery semantics (v0.4.0+)

​OnQuarantined - quarantine semantics (v0.4.0+)

​Threading contract

​Counter pattern

​Active-connection gauge pattern

​Per-event-type counter

​Independence from Logger

​Compatibility note

The interface

Hook semantics

What counts as a “whitelisted event”

`OnFallback` reasons

`FallbackReasonRateLimited` - active-recovery semantics (v0.4.0+)

`OnQuarantined` - quarantine semantics (v0.4.0+)

Threading contract

Counter pattern

Active-connection gauge pattern

Per-event-type counter

Independence from `Logger`

Compatibility note