From 2319aeb8fd456bdb98f90286bffe786d58cde426 Mon Sep 17 00:00:00 2001 From: Jamie Sinn Date: Tue, 4 Aug 2026 16:50:02 -0400 Subject: [PATCH 1/9] feat: SSE support for OFREP endpoints Signed-off-by: Jamie Sinn --- core/pkg/service/ofrep/models.go | 22 +- core/pkg/store/query.go | 11 + flagd/cmd/start.go | 12 + flagd/go.mod | 1 + flagd/go.sum | 4 + flagd/pkg/runtime/from_config.go | 8 +- .../service/flag-evaluation/ofrep/handler.go | 98 ++++++- .../flag-evaluation/ofrep/ofrep_service.go | 48 +++- .../ofrep/ofrep_service_test.go | 4 +- .../flag-evaluation/ofrep/sse/event.go | 36 +++ .../flag-evaluation/ofrep/sse/handler.go | 50 ++++ .../flag-evaluation/ofrep/sse/service.go | 91 +++++++ .../flag-evaluation/ofrep/sse/service_test.go | 53 ++++ .../flag-evaluation/ofrep/sse/tracker.go | 247 ++++++++++++++++++ .../flag-evaluation/ofrep/sse/tracker_test.go | 111 ++++++++ .../flag-evaluation/ofrep/sse_bulk_test.go | 157 +++++++++++ test/zero-downtime-flagd-proxy/go.sum | 12 +- 17 files changed, 948 insertions(+), 17 deletions(-) create mode 100644 flagd/pkg/service/flag-evaluation/ofrep/sse/event.go create mode 100644 flagd/pkg/service/flag-evaluation/ofrep/sse/handler.go create mode 100644 flagd/pkg/service/flag-evaluation/ofrep/sse/service.go create mode 100644 flagd/pkg/service/flag-evaluation/ofrep/sse/service_test.go create mode 100644 flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go create mode 100644 flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go create mode 100644 flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go diff --git a/core/pkg/service/ofrep/models.go b/core/pkg/service/ofrep/models.go index e7ff02849..2f656ae2e 100644 --- a/core/pkg/service/ofrep/models.go +++ b/core/pkg/service/ofrep/models.go @@ -22,6 +22,24 @@ type EvaluationSuccess struct { type BulkEvaluationResponse struct { Flags []interface{} `json:"flags"` Metadata model.Metadata `json:"metadata"` + // EventStreams advertises SSE endpoints clients can subscribe to for change + // notifications, per OpenFeature protocol ADR-0008. Omitted when SSE is disabled. + EventStreams []EventStream `json:"eventStreams,omitempty"` +} + +// EventStream describes a Server-Sent Events endpoint a client can subscribe to in order to +// be notified (via a `refetchEvaluation` event) when the flag configuration changes. +type EventStream struct { + Type string `json:"type"` + Endpoint *EventStreamEndpoint `json:"endpoint"` + InactivityDelaySec int `json:"inactivityDelaySec,omitempty"` +} + +// EventStreamEndpoint is the ADR-0008 structured form of an event-stream location. Origin is +// optional; when omitted the client resolves RequestUri against its OFREP base URL origin. +type EventStreamEndpoint struct { + Origin string `json:"origin,omitempty"` + RequestUri string `json:"requestUri"` } type EvaluationError struct { @@ -54,8 +72,8 @@ func BulkEvaluationResponseFrom(resolutions []evaluator.AnyValue, metadata model } return BulkEvaluationResponse{ - evaluations, - metadata, + Flags: evaluations, + Metadata: metadata, } } diff --git a/core/pkg/store/query.go b/core/pkg/store/query.go index 22dcaa558..d62d19c22 100644 --- a/core/pkg/store/query.go +++ b/core/pkg/store/query.go @@ -28,6 +28,11 @@ const flagSetIdKeySourceCompoundIndex = flagSetIdIndex + "+" + keyIndex + "+" + // any flag without a "flagSetId" is assigned this one; it's never exposed externally var nilFlagSetId = uuid.New().String() +// NilFlagSetId returns the startup-generated flagSetId assigned to flags that have +// no explicit flagSetId. It is never exposed to clients, but consumers that group or +// fingerprint flags by flagSetId (e.g. the OFREP SSE tracker) need to recognise it. +func NilFlagSetId() string { return nilFlagSetId } + // A Selector represents a set of constraints used to query the store. type Selector struct { indexMap map[string]string @@ -83,6 +88,12 @@ func (s Selector) WithSource(source string) Selector { return s.withIndex(source func (s Selector) WithFlagSetId(id string) Selector { return s.withIndex(flagSetIdIndex, id) } func (s Selector) withKey(key string) Selector { return s.withIndex(keyIndex, key) } +// FlagSetId returns the flagSetId constraint of the selector, or "" if none is set. +func (s Selector) FlagSetId() string { return s.indexMap[flagSetIdIndex] } + +// Source returns the source constraint of the selector, or "" if none is set. +func (s Selector) Source() string { return s.indexMap[sourceIndex] } + func (s Selector) withIndex(key, value string) Selector { m := maps.Clone(s.indexMap) if m == nil { diff --git a/flagd/cmd/start.go b/flagd/cmd/start.go index e498304e1..b6de8aefc 100644 --- a/flagd/cmd/start.go +++ b/flagd/cmd/start.go @@ -23,6 +23,9 @@ const ( managementPortFlagName = "management-port" metricsExporter = "metrics-exporter" ofrepPortFlagName = "ofrep-port" + ofrepSSEEnabledFlagName = "ofrep-sse-enabled" + ofrepSSEInactivityFlagName = "ofrep-sse-inactivity-delay" + ofrepSSEPublicURLFlagName = "ofrep-sse-public-url" otelCollectorURI = "otel-collector-uri" otelCertPathFlagName = "otel-cert-path" otelKeyPathFlagName = "otel-key-path" @@ -58,6 +61,9 @@ func init() { flags.Int32P(syncPortFlagName, "g", 8015, "gRPC Sync port") flags.Int32P(ofrepPortFlagName, "r", 8016, "ofrep service port") + flags.Bool(ofrepSSEEnabledFlagName, true, "Enable the OFREP SSE change-notification endpoint (ADR-0008) at /ofrep/v1/sse on the ofrep port. Defaults to true.") + flags.Int(ofrepSSEInactivityFlagName, 120, "Inactivity delay (seconds) advertised to OFREP SSE clients in the eventStreams block. Clients close idle connections after this. Defaults to 120.") + flags.String(ofrepSSEPublicURLFlagName, "", "Origin (scheme://host) advertised as the OFREP SSE eventStreams endpoint.origin. Omitted when empty, so clients resolve the requestUri against the OFREP base URL. Set when flagd is behind a proxy.") flags.StringP(socketPathFlagName, "d", "", "Flagd unix socket path. "+ "With grpc the evaluations service will become available on this address. "+ "With http(s) the grpc-gateway proxy will use this address internally.") @@ -123,6 +129,9 @@ func bindFlags(flags *pflag.FlagSet) { _ = viper.BindPFlag(syncPortFlagName, flags.Lookup(syncPortFlagName)) _ = viper.BindPFlag(syncSocketPathFlagName, flags.Lookup(syncSocketPathFlagName)) _ = viper.BindPFlag(ofrepPortFlagName, flags.Lookup(ofrepPortFlagName)) + _ = viper.BindPFlag(ofrepSSEEnabledFlagName, flags.Lookup(ofrepSSEEnabledFlagName)) + _ = viper.BindPFlag(ofrepSSEInactivityFlagName, flags.Lookup(ofrepSSEInactivityFlagName)) + _ = viper.BindPFlag(ofrepSSEPublicURLFlagName, flags.Lookup(ofrepSSEPublicURLFlagName)) _ = viper.BindPFlag(contextValueFlagName, flags.Lookup(contextValueFlagName)) _ = viper.BindPFlag(headerToContextKeyFlagName, flags.Lookup(headerToContextKeyFlagName)) _ = viper.BindPFlag(streamDeadlineFlagName, flags.Lookup(streamDeadlineFlagName)) @@ -201,6 +210,9 @@ var startCmd = &cobra.Command{ MetricExporter: viper.GetString(metricsExporter), ManagementPort: viper.GetUint16(managementPortFlagName), OfrepServicePort: viper.GetUint16(ofrepPortFlagName), + OfrepSSEEnabled: viper.GetBool(ofrepSSEEnabledFlagName), + OfrepSSEInactivityDel: viper.GetInt(ofrepSSEInactivityFlagName), + OfrepSSEPublicURL: viper.GetString(ofrepSSEPublicURLFlagName), OtelCollectorURI: viper.GetString(otelCollectorURI), OtelCertPath: viper.GetString(otelCertPathFlagName), OtelKeyPath: viper.GetString(otelKeyPathFlagName), diff --git a/flagd/go.mod b/flagd/go.mod index 3ce5f74af..6f40cebfd 100644 --- a/flagd/go.mod +++ b/flagd/go.mod @@ -9,6 +9,7 @@ require ( connectrpc.com/connect v1.19.1 github.com/dimiro1/banner v1.1.0 github.com/gorilla/mux v1.8.1 + github.com/launchdarkly/eventsource v1.11.0 github.com/mattn/go-colorable v0.1.14 github.com/open-feature/flagd/core v0.15.6 github.com/prometheus/client_golang v1.23.2 diff --git a/flagd/go.sum b/flagd/go.sum index b1a4529cd..fd196ad2a 100644 --- a/flagd/go.sum +++ b/flagd/go.sum @@ -255,6 +255,10 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= +github.com/launchdarkly/eventsource v1.11.0 h1:aAdvh2XmtXA17QsRFL0XKHURMqhxg7J+CceQmhSzBas= +github.com/launchdarkly/eventsource v1.11.0/go.mod h1:dU+rZxkPOlGPsyJPpiDqiepAcFwIITDUClY9+A6RrMw= +github.com/launchdarkly/go-test-helpers/v3 v3.1.0 h1:E3bxJMzMoA+cJSF3xxtk2/chr1zshl1ZWa0/oR+8bvg= +github.com/launchdarkly/go-test-helpers/v3 v3.1.0/go.mod h1:Ake5+hZFS/DmIGKx/cizhn5W9pGA7pplcR7xCxWiLIo= github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0= github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc= github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE= diff --git a/flagd/pkg/runtime/from_config.go b/flagd/pkg/runtime/from_config.go index 777a1af98..7b93e1011 100644 --- a/flagd/pkg/runtime/from_config.go +++ b/flagd/pkg/runtime/from_config.go @@ -27,6 +27,9 @@ type Config struct { MetricExporter string ManagementPort uint16 OfrepServicePort uint16 + OfrepSSEEnabled bool + OfrepSSEInactivityDel int + OfrepSSEPublicURL string OtelCollectorURI string OtelCertPath string OtelKeyPath string @@ -112,13 +115,16 @@ func FromConfig(logger *logger.Logger, version string, config Config) (*Runtime, recorder) // ofrep service - ofrepService, err := ofrep.NewOfrepService(jsonEvaluator, config.CORS, ofrep.SvcConfiguration{ + ofrepService, err := ofrep.NewOfrepService(jsonEvaluator, store, config.CORS, ofrep.SvcConfiguration{ Logger: logger.WithFields(zap.String("component", "OFREPService")), Port: config.OfrepServicePort, ServiceName: svcName, MetricsRecorder: recorder, MaxRequestBodyBytes: config.MaxRequestBodyBytes, MaxRequestHeaderBytes: config.MaxRequestHeaderBytes, + SSEEnabled: config.OfrepSSEEnabled, + SSEInactivityDelaySec: config.OfrepSSEInactivityDel, + SSEPublicURL: config.OfrepSSEPublicURL, }, config.ContextValues, config.HeaderToContextKeyMappings, diff --git a/flagd/pkg/service/flag-evaluation/ofrep/handler.go b/flagd/pkg/service/flag-evaluation/ofrep/handler.go index bd4b8a344..484285d21 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/handler.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/handler.go @@ -6,6 +6,8 @@ import ( "errors" "fmt" "net/http" + "net/url" + "strings" "github.com/gorilla/mux" "github.com/open-feature/flagd/core/pkg/evaluator" @@ -29,6 +31,13 @@ const ( bulkEvaluation = "/ofrep/v1/evaluate/{path:flags\\/|flags}" ) +// configVersioner resolves the current config ETag / last-modified time for a selector so the +// bulk handler can serve conditional (ETag/304) responses consistent with the SSE stream. +// Implemented by the OFREP SSE change tracker; nil when SSE is disabled. +type configVersioner interface { + Version(selector store.Selector) (etag string, lastModified int64, ok bool) +} + type handler struct { Logger *logger.Logger evaluator evaluator.IEvaluator @@ -36,6 +45,20 @@ type handler struct { headerToContextKeyMappings map[string]string metricsRecorder telemetry.IMetricsRecorder tracer trace.Tracer + + versioner configVersioner + sseEnabled bool + sseInactivityDelaySec int + ssePublicURL string +} + +// SSEConfig carries the SSE advertisement settings the bulk handler needs to expose the +// `eventStreams` block and conditional-evaluation ETags. +type SSEConfig struct { + Enabled bool + Versioner configVersioner + InactivityDelaySec int + PublicURL string } func NewOfrepHandler( @@ -45,6 +68,7 @@ func NewOfrepHandler( headerToContextKeyMappings map[string]string, metricsRecorder telemetry.IMetricsRecorder, serviceName string, + sseCfg SSEConfig, ) http.Handler { h := handler{ Logger: logger, @@ -53,6 +77,10 @@ func NewOfrepHandler( headerToContextKeyMappings: headerToContextKeyMappings, metricsRecorder: metricsRecorder, tracer: otel.Tracer("flagd.ofrep.v1"), + versioner: sseCfg.Versioner, + sseEnabled: sseCfg.Enabled, + sseInactivityDelaySec: sseCfg.InactivityDelaySec, + ssePublicURL: sseCfg.PublicURL, } router := mux.NewRouter() @@ -143,6 +171,20 @@ func (h *handler) HandleBulkEvaluation(w http.ResponseWriter, r *http.Request) { } ctx := context.WithValue(r.Context(), store.SelectorContextKey{}, selector) + // Conditional evaluation (ADR-0008): if the client already holds the current config + // version, short-circuit with 304 Not Modified instead of re-serving the flags. + etag, lastModified, hasVersion := "", int64(0), false + if h.versioner != nil { + etag, lastModified, hasVersion = h.versioner.Version(selector) + } + if hasVersion && etag != "" { + w.Header().Set("ETag", quoteETag(etag)) + if clientEtag := requestETag(r); clientEtag != "" && normalizeETag(clientEtag) == etag { + w.WriteHeader(http.StatusNotModified) + return + } + } + evaluations, metadata, err := h.evaluator.ResolveAllValues(ctx, requestID, evaluationContext) if h.metricsRecorder != nil { for _, evaluation := range evaluations { @@ -156,8 +198,62 @@ func (h *handler) HandleBulkEvaluation(w http.ResponseWriter, r *http.Request) { fmt.Sprintf("Bulk evaluation failed. Tracking ID: %s", requestID)) h.writeJSONToResponse(http.StatusInternalServerError, res, w) } else { - h.writeJSONToResponse(http.StatusOK, ofrep.BulkEvaluationResponseFrom(evaluations, metadata), w) + response := ofrep.BulkEvaluationResponseFrom(evaluations, metadata) + if h.sseEnabled { + response.EventStreams = h.eventStreams(selector) + if lastModified > 0 { + if response.Metadata == nil { + response.Metadata = model.Metadata{} + } + response.Metadata["flagConfigLastModified"] = lastModified + } + } + h.writeJSONToResponse(http.StatusOK, response, w) + } +} + +// eventStreams builds the ADR-0008 eventStreams advertisement pointing OFREP clients back at +// this flagd's SSE endpoint. The channel is the request's flagSetId; the internal nilFlagSetId +// and a missing flagSetId both map to the catch-all channel (no `channels` parameter). +// +// It uses the structured `endpoint` form and omits origin unless a public URL is configured, so +// the client resolves the requestUri against the OFREP base URL it is already talking to. +func (h *handler) eventStreams(selector store.Selector) []ofrep.EventStream { + channel := selector.FlagSetId() + if channel == store.NilFlagSetId() { + channel = "" + } + + requestURI := ssePath + if channel != "" { + requestURI += "?channels=" + url.QueryEscape(channel) + } + + return []ofrep.EventStream{{ + Type: "sse", + InactivityDelaySec: h.sseInactivityDelaySec, + Endpoint: &ofrep.EventStreamEndpoint{ + Origin: strings.TrimSuffix(h.ssePublicURL, "/"), + RequestUri: requestURI, + }, + }} +} + +func requestETag(r *http.Request) string { + if e := r.URL.Query().Get("flagConfigEtag"); e != "" { + return e } + return r.Header.Get("If-None-Match") +} + +// normalizeETag strips optional surrounding quotes so quoted and unquoted forms compare equal. +func normalizeETag(etag string) string { + return strings.Trim(etag, `"`) +} + +// quoteETag wraps a bare ETag value in the double quotes required by the HTTP ETag header. +func quoteETag(etag string) string { + return `"` + etag + `"` } func (h *handler) writeJSONToResponse(status int, payload interface{}, w http.ResponseWriter) { diff --git a/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service.go b/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service.go index 9a8994469..3ce9064ad 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service.go @@ -9,11 +9,16 @@ import ( "github.com/open-feature/flagd/core/pkg/evaluator" "github.com/open-feature/flagd/core/pkg/logger" + "github.com/open-feature/flagd/core/pkg/store" "github.com/open-feature/flagd/core/pkg/telemetry" + "github.com/open-feature/flagd/flagd/pkg/service/flag-evaluation/ofrep/sse" corsmw "github.com/open-feature/flagd/flagd/pkg/service/middleware/cors" "golang.org/x/sync/errgroup" ) +// ssePath is the endpoint OFREP clients subscribe to for change notifications (ADR-0008). +const ssePath = "/ofrep/v1/sse" + type IOfrepService interface { // Start the OFREP service with context for shutdown Start(context.Context) error @@ -26,30 +31,60 @@ type SvcConfiguration struct { MetricsRecorder telemetry.IMetricsRecorder MaxRequestBodyBytes int64 MaxRequestHeaderBytes int64 + + // SSE (ADR-0008) settings + SSEEnabled bool + SSEInactivityDelaySec int + SSEPublicURL string } type Service struct { logger *logger.Logger port uint16 server *http.Server + sse *sse.Service } func NewOfrepService( - evaluator evaluator.IEvaluator, origins []string, cfg SvcConfiguration, contextValues map[string]any, headerToContextKeyMappings map[string]string, + evaluator evaluator.IEvaluator, flagStore store.IStore, origins []string, cfg SvcConfiguration, contextValues map[string]any, headerToContextKeyMappings map[string]string, ) (*Service, error) { corsMiddleware := corsmw.New(origins) - var h http.Handler = NewOfrepHandler( + var sseService *sse.Service + sseCfg := SSEConfig{ + Enabled: cfg.SSEEnabled, + InactivityDelaySec: cfg.SSEInactivityDelaySec, + PublicURL: cfg.SSEPublicURL, + } + if cfg.SSEEnabled { + sseService = sse.New(flagStore, sse.Config{Logger: cfg.Logger}) + sseCfg.Versioner = sseService.Tracker() + } + + ofrepHandler := NewOfrepHandler( cfg.Logger, evaluator, contextValues, headerToContextKeyMappings, cfg.MetricsRecorder, cfg.ServiceName, + sseCfg, ) + + // Route the long-lived SSE stream separately from the request/response evaluate routes so + // the request-body limit is not applied to the stream. Everything else falls through to the + // existing OFREP handler. + mux := http.NewServeMux() + var evaluateHandler http.Handler = ofrepHandler if cfg.MaxRequestBodyBytes > 0 { - h = http.MaxBytesHandler(h, cfg.MaxRequestBodyBytes) + evaluateHandler = http.MaxBytesHandler(evaluateHandler, cfg.MaxRequestBodyBytes) + } + if sseService != nil { + mux.Handle(ssePath, sseService.Handler()) } + mux.Handle("/", evaluateHandler) + + var h http.Handler = mux h = corsMiddleware.Handler(h) server := http.Server{ @@ -65,6 +100,7 @@ func NewOfrepService( logger: cfg.Logger, port: cfg.Port, server: &server, + sse: sseService, }, nil } @@ -81,6 +117,12 @@ func (s Service) Start(ctx context.Context) error { return nil }) + if s.sse != nil { + group.Go(func() error { + return s.sse.Start(gCtx) + }) + } + group.Go(func() error { <-gCtx.Done() s.logger.Info("shutting down ofrep service") diff --git a/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service_test.go b/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service_test.go index 90682a4c4..51e2ed462 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service_test.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service_test.go @@ -37,7 +37,7 @@ func TestOfrepServiceStartStop(t *testing.T) { MetricsRecorder: &telemetry.NoopMetricsRecorder{}, } - service, err := NewOfrepService(eval, []string{"*"}, cfg, nil, nil) + service, err := NewOfrepService(eval, nil, []string{"*"}, cfg, nil, nil) if err != nil { t.Fatalf(errCreateOfrepService, err) } @@ -193,7 +193,7 @@ func startOfrepService(t *testing.T, cfg SvcConfiguration) (*Service, uint16) { t.Helper() eval := mock.NewMockIEvaluator(gomock.NewController(t)) - service, err := NewOfrepService(eval, []string{"*"}, cfg, nil, nil) + service, err := NewOfrepService(eval, nil, []string{"*"}, cfg, nil, nil) if err != nil { t.Fatalf(errCreateOfrepService, err) } diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/event.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/event.go new file mode 100644 index 000000000..811402d74 --- /dev/null +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/event.go @@ -0,0 +1,36 @@ +package sse + +import "encoding/json" + +// refetchEventType is the only event data type defined by OpenFeature protocol ADR-0008. +const refetchEventType = "refetchEvaluation" + +// eventName is the SSE `event:` field, using the ADR-0008 "message" envelope. +const eventName = "message" + +type refetchPayload struct { + Type string `json:"type"` + Etag string `json:"etag,omitempty"` + LastModified int64 `json:"lastModified,omitempty"` +} + +// refetchEvent implements eventsource.Event, telling subscribed OFREP clients that the flag +// configuration for their channel changed and they should re-run bulk evaluation. +type refetchEvent struct { + id string + data string +} + +func newRefetchEvent(id, etag string, lastModified int64) refetchEvent { + payload := refetchPayload{ + Type: refetchEventType, + Etag: etag, + LastModified: lastModified, + } + data, _ := json.Marshal(payload) + return refetchEvent{id: id, data: string(data)} +} + +func (e refetchEvent) Id() string { return e.id } +func (e refetchEvent) Event() string { return eventName } +func (e refetchEvent) Data() string { return e.data } diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/handler.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/handler.go new file mode 100644 index 000000000..44c51e97c --- /dev/null +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/handler.go @@ -0,0 +1,50 @@ +package sse + +import ( + "net/http" + "sync" +) + +// channelParam carries the ADR-0008 channel token (the flagSetId); empty selects the catch-all. +const channelParam = "channels" + +func channelFromRequest(r *http.Request) string { + return r.URL.Query().Get(channelParam) +} + +// activeChannels is a reference-counted set of channels with at least one live subscriber, so +// the heartbeat loop knows which channels to ping (the eventsource server exposes no registry). +type activeChannels struct { + mu sync.Mutex + counts map[string]int +} + +func newActiveChannels() *activeChannels { + return &activeChannels{counts: map[string]int{}} +} + +func (a *activeChannels) add(channel string) { + a.mu.Lock() + defer a.mu.Unlock() + a.counts[channel]++ +} + +func (a *activeChannels) remove(channel string) { + a.mu.Lock() + defer a.mu.Unlock() + if a.counts[channel] <= 1 { + delete(a.counts, channel) + return + } + a.counts[channel]-- +} + +func (a *activeChannels) snapshot() []string { + a.mu.Lock() + defer a.mu.Unlock() + channels := make([]string, 0, len(a.counts)) + for ch := range a.counts { + channels = append(channels, ch) + } + return channels +} diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/service.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/service.go new file mode 100644 index 000000000..97109b7b0 --- /dev/null +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/service.go @@ -0,0 +1,91 @@ +package sse + +import ( + "context" + "net/http" + "time" + + "github.com/launchdarkly/eventsource" + "github.com/open-feature/flagd/core/pkg/logger" + "github.com/open-feature/flagd/core/pkg/store" +) + +// defaultHeartbeatInterval must stay well below the advertised inactivity delay so idle +// connections are not closed by clients or proxies. +const defaultHeartbeatInterval = 30 * time.Second + +type Config struct { + Logger *logger.Logger + HeartbeatInterval time.Duration +} + +// Service serves the OFREP SSE stream. It owns an eventsource server, the change Tracker that +// drives it, and a heartbeat loop that keeps idle connections alive. +type Service struct { + logger *logger.Logger + es *eventsource.Server + tracker *Tracker + active *activeChannels + heartbeatInterval time.Duration +} + +// New builds an SSE Service backed by the shared flag store. +func New(s store.IStore, cfg Config) *Service { + es := eventsource.NewServer() + // CORS is handled by the OFREP server's middleware; avoid emitting duplicate headers. + es.AllowCORS = false + es.ReplayAll = false + + heartbeat := cfg.HeartbeatInterval + if heartbeat <= 0 { + heartbeat = defaultHeartbeatInterval + } + + return &Service{ + logger: cfg.Logger, + es: es, + tracker: NewTracker(cfg.Logger, s, es), + active: newActiveChannels(), + heartbeatInterval: heartbeat, + } +} + +// Tracker exposes the change tracker so the OFREP bulk handler can resolve config versions +// for conditional (ETag/304) evaluation. +func (svc *Service) Tracker() *Tracker { return svc.tracker } + +// Handler registers the request's channel in the active set (for heartbeats) and delegates to +// the eventsource server, which streams until the client disconnects. +func (svc *Service) Handler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + channel := channelFromRequest(r) + svc.active.add(channel) + defer svc.active.remove(channel) + svc.es.Handler(channel).ServeHTTP(w, r) + }) +} + +// Start runs the change tracker and heartbeat loop until ctx is cancelled, then shuts the +// eventsource server down. It blocks and is intended to run in its own goroutine. +func (svc *Service) Start(ctx context.Context) error { + go svc.tracker.Run(ctx) + + ticker := time.NewTicker(svc.heartbeatInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + svc.es.Close() + if svc.logger != nil { + svc.logger.Info("shutting down ofrep sse service") + } + return nil + case <-ticker.C: + channels := svc.active.snapshot() + if len(channels) > 0 { + svc.es.PublishComment(channels, "keep-alive") + } + } + } +} diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/service_test.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/service_test.go new file mode 100644 index 000000000..be62a9b9f --- /dev/null +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/service_test.go @@ -0,0 +1,53 @@ +package sse + +import ( + "context" + "net/http/httptest" + "testing" + "time" + + "github.com/launchdarkly/eventsource" + "github.com/open-feature/flagd/core/pkg/logger" + "github.com/open-feature/flagd/core/pkg/model" + "github.com/open-feature/flagd/core/pkg/store" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestService_PublishesRefetchOnChange verifies the end-to-end wiring: a client subscribed to a +// flagSetId channel receives an ADR-0008 refetchEvaluation event when that flag set changes. +func TestService_PublishesRefetchOnChange(t *testing.T) { + log := logger.NewLogger(nil, false) + s, err := store.NewStore(log, []string{"src1"}) + require.NoError(t, err) + + // long heartbeat so keep-alive comments do not interfere with the assertion + svc := New(s, Config{Logger: log, HeartbeatInterval: time.Hour}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = svc.Start(ctx) }() + + ts := httptest.NewServer(svc.Handler()) + defer ts.Close() + + // allow the tracker's initial (empty) snapshot to be consumed and skipped + time.Sleep(100 * time.Millisecond) + + stream, err := eventsource.Subscribe(ts.URL+"?channels=fs1", "") + require.NoError(t, err) + defer stream.Close() + + // allow the subscription to register server-side before publishing + time.Sleep(100 * time.Millisecond) + + s.Update("src1", []model.Flag{testFlag("fs1", "a", "on")}, model.Metadata{"flagSetId": "fs1"}, false) + + select { + case ev := <-stream.Events: + assert.Equal(t, eventName, ev.Event()) + assert.Contains(t, ev.Data(), refetchEventType) + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for refetch event on channel fs1") + } +} diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go new file mode 100644 index 000000000..a0bd21889 --- /dev/null +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go @@ -0,0 +1,247 @@ +package sse + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/launchdarkly/eventsource" + "github.com/open-feature/flagd/core/pkg/logger" + "github.com/open-feature/flagd/core/pkg/model" + "github.com/open-feature/flagd/core/pkg/store" +) + +// allChannel receives a refetch event on any flag change. It is advertised to OFREP clients +// that use no flagSetId selector, mirroring a bulk request with no selector. +const allChannel = "" + +// namespacing prefixes for the internal version map, so a flagSetId and a source with the +// same string value never collide. +const ( + allKey = "all" + fsPrefix = "fs:" + srcPrefix = "src:" +) + +func fsKey(id string) string { return fsPrefix + id } +func srcKey(src string) string { return srcPrefix + src } + +type version struct { + etag string + lastModified int64 +} + +// Tracker owns a global store.Watch subscription. On every flag-configuration change it +// recomputes a per-channel config fingerprint (used as an ETag), publishes an ADR-0008 +// refetchEvaluation event to each affected channel, and answers version lookups for the OFREP +// bulk handler so conditional (ETag/304) evaluation stays consistent with the SSE stream. +type Tracker struct { + logger *logger.Logger + store store.IStore + es *eventsource.Server + + mu sync.RWMutex + versions map[string]version + + eventID atomic.Int64 +} + +// NewTracker creates a Tracker. Call Run to begin watching the store. +func NewTracker(log *logger.Logger, s store.IStore, es *eventsource.Server) *Tracker { + return &Tracker{ + logger: log, + store: s, + es: es, + versions: map[string]version{}, + } +} + +// Run subscribes to all flag changes and publishes refetch events until ctx is cancelled. +// It blocks, so it is intended to run in its own goroutine. +func (t *Tracker) Run(ctx context.Context) { + watcher := make(chan store.FlagQueryResult, 1) + t.store.Watch(ctx, &store.Selector{}, watcher) + + first := true + for res := range watcher { + channels := t.update(res.Flags) + if first { + // The first emission is the initialization snapshot; only seed fingerprints. + first = false + continue + } + for _, ch := range channels { + t.publish(ch) + } + } +} + +// Version returns the current config ETag and last-modified time (unix seconds) for the +// channel matching the given selector. ok is false when no version is tracked for the +// selector (e.g. an unknown source), letting the caller skip conditional handling. +func (t *Tracker) Version(selector store.Selector) (etag string, lastModified int64, ok bool) { + var key string + switch { + case selector.IsEmpty(): + key = allKey + case selector.FlagSetId() != "": + key = fsKey(selector.FlagSetId()) + case selector.Source() != "": + key = srcKey(selector.Source()) + default: + return "", 0, false + } + + t.mu.RLock() + defer t.mu.RUnlock() + v, ok := t.versions[key] + return v.etag, v.lastModified, ok +} + +// update recomputes fingerprints for the catch-all, per-flagSetId and per-source groups, +// swaps them into the version map (preserving lastModified when a fingerprint is unchanged) +// and returns the eventsource channels that should be notified. +func (t *Tracker) update(flags []model.Flag) []string { + now := time.Now().Unix() + + fsGroups := map[string][]model.Flag{} + srcGroups := map[string][]model.Flag{} + for _, f := range flags { + fsGroups[f.FlagSetId] = append(fsGroups[f.FlagSetId], f) + srcGroups[f.Source] = append(srcGroups[f.Source], f) + } + + newVersions := make(map[string]version, len(fsGroups)+len(srcGroups)+1) + newVersions[allKey] = version{etag: fingerprint(flags)} + for id, g := range fsGroups { + newVersions[fsKey(id)] = version{etag: fingerprint(g)} + } + for src, g := range srcGroups { + newVersions[srcKey(src)] = version{etag: fingerprint(g)} + } + + t.mu.Lock() + defer t.mu.Unlock() + + // carry lastModified forward when the fingerprint is unchanged + for key, nv := range newVersions { + if old, exists := t.versions[key]; exists && old.etag == nv.etag { + nv.lastModified = old.lastModified + } else { + nv.lastModified = now + } + newVersions[key] = nv + } + + changed := t.changedChannels(newVersions, fsGroups) + t.versions = newVersions + return changed +} + +// changedChannels compares the previous version map (t.versions, still held) with the freshly +// computed one and returns the eventsource channels that clients subscribe to and whose config +// changed: the catch-all channel plus any created/updated/removed flagSetId channels. Source +// channels are tracked for ETag lookups but are not (yet) directly subscribable. +func (t *Tracker) changedChannels(newVersions map[string]version, fsGroups map[string][]model.Flag) []string { + etagChanged := func(key string) bool { + old, oldOK := t.versions[key] + nv, newOK := newVersions[key] + if oldOK != newOK { + return true + } + return old.etag != nv.etag + } + + notify := map[string]struct{}{} + if etagChanged(allKey) { + notify[allChannel] = struct{}{} + } + + nilID := store.NilFlagSetId() + for id := range fsGroups { + if id == nilID { + continue // internal flagSetId, never subscribable + } + if etagChanged(fsKey(id)) { + notify[id] = struct{}{} + } + } + // flagSetId channels that existed before but have no flags now (whole set deleted) + for key := range t.versions { + id, ok := strings.CutPrefix(key, fsPrefix) + if !ok || id == nilID { + continue + } + if _, present := fsGroups[id]; !present { + notify[id] = struct{}{} + } + } + + channels := make([]string, 0, len(notify)) + for ch := range notify { + channels = append(channels, ch) + } + return channels +} + +// publish emits a refetch event to a single eventsource channel using that channel's current +// ETag and lastModified. +func (t *Tracker) publish(channel string) { + key := allKey + if channel != allChannel { + key = fsKey(channel) + } + + t.mu.RLock() + v := t.versions[key] + t.mu.RUnlock() + + id := strconv.FormatInt(t.eventID.Add(1), 10) + t.es.Publish([]string{channel}, newRefetchEvent(id, v.etag, v.lastModified)) + if t.logger != nil { + t.logger.Debug(fmt.Sprintf("published refetch event to channel %q (etag=%s)", channel, v.etag)) + } +} + +// fingerprint produces a deterministic, restart-stable hash of a group of flag definitions. +// nilFlagSetId is normalised to "" so identical config yields the same fingerprint across +// restarts. +func fingerprint(flags []model.Flag) string { + sorted := make([]model.Flag, len(flags)) + copy(sorted, flags) + sort.Slice(sorted, func(i, j int) bool { + if sorted[i].FlagSetId != sorted[j].FlagSetId { + return sorted[i].FlagSetId < sorted[j].FlagSetId + } + if sorted[i].Source != sorted[j].Source { + return sorted[i].Source < sorted[j].Source + } + return sorted[i].Key < sorted[j].Key + }) + + nilID := store.NilFlagSetId() + h := sha256.New() + for _, f := range sorted { + fsid := f.FlagSetId + if fsid == nilID { + fsid = "" + } + h.Write([]byte(fsid)) + h.Write([]byte{0}) + h.Write([]byte(f.Key)) + h.Write([]byte{0}) + // model.Flag.MarshalJSON is stable (map keys sorted) and covers the definition fields. + b, _ := json.Marshal(f) + h.Write(b) + h.Write([]byte{0}) + } + return hex.EncodeToString(h.Sum(nil)) +} diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go new file mode 100644 index 000000000..580397767 --- /dev/null +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go @@ -0,0 +1,111 @@ +package sse + +import ( + "testing" + + "github.com/open-feature/flagd/core/pkg/model" + "github.com/open-feature/flagd/core/pkg/store" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func testFlag(flagSetID, key, defaultVariant string) model.Flag { + return model.Flag{ + Key: key, + FlagSetId: flagSetID, + State: "ENABLED", + DefaultVariant: defaultVariant, + Variants: map[string]any{"on": true, "off": false}, + Source: "src1", + } +} + +func mustSelector(t *testing.T, expr string) store.Selector { + t.Helper() + s, err := store.NewSelector(expr) + require.NoError(t, err) + return s +} + +func TestFingerprint_StableAndSensitive(t *testing.T) { + a := testFlag("fs1", "a", "on") + b := testFlag("fs1", "b", "off") + + // deterministic for identical input + assert.Equal(t, fingerprint([]model.Flag{a, b}), fingerprint([]model.Flag{a, b})) + // order independent + assert.Equal(t, fingerprint([]model.Flag{a, b}), fingerprint([]model.Flag{b, a})) + + // sensitive to a definition change + aModified := testFlag("fs1", "a", "off") + assert.NotEqual(t, fingerprint([]model.Flag{a}), fingerprint([]model.Flag{aModified})) + + // nilFlagSetId is normalised so it does not leak the random UUID into the hash + nilGroup := fingerprint([]model.Flag{testFlag(store.NilFlagSetId(), "a", "on")}) + emptyGroup := fingerprint([]model.Flag{testFlag("", "a", "on")}) + assert.Equal(t, emptyGroup, nilGroup) +} + +func TestTracker_Update_ChangedChannels(t *testing.T) { + tr := &Tracker{versions: map[string]version{}} + + fs1a := testFlag("fs1", "a", "on") + fs2b := testFlag("fs2", "b", "on") + + // first update seeds every channel (this is the init snapshot Run skips) + first := tr.update([]model.Flag{fs1a, fs2b}) + assert.Contains(t, first, allChannel) + assert.Contains(t, first, "fs1") + assert.Contains(t, first, "fs2") + + // no change -> no channels + assert.Empty(t, tr.update([]model.Flag{fs1a, fs2b})) + + // change only fs1 -> catch-all + fs1, not fs2 + fs1aModified := testFlag("fs1", "a", "off") + changed := tr.update([]model.Flag{fs1aModified, fs2b}) + assert.Contains(t, changed, allChannel) + assert.Contains(t, changed, "fs1") + assert.NotContains(t, changed, "fs2") + + // deleting a whole flag set notifies that set's channel + catch-all + removed := tr.update([]model.Flag{fs1aModified}) + assert.Contains(t, removed, allChannel) + assert.Contains(t, removed, "fs2") + assert.NotContains(t, removed, "fs1") +} + +func TestTracker_Update_NilFlagSetIdNotSubscribable(t *testing.T) { + tr := &Tracker{versions: map[string]version{}} + + changed := tr.update([]model.Flag{testFlag(store.NilFlagSetId(), "a", "on")}) + assert.Contains(t, changed, allChannel, "catch-all must fire for flags without a flagSetId") + assert.NotContains(t, changed, store.NilFlagSetId(), "internal nilFlagSetId must not be a subscribable channel") +} + +func TestTracker_Version(t *testing.T) { + tr := &Tracker{versions: map[string]version{}} + tr.update([]model.Flag{testFlag("fs1", "a", "on")}) + + // catch-all (empty selector) + allEtag, _, ok := tr.Version(store.Selector{}) + require.True(t, ok) + assert.NotEmpty(t, allEtag) + + // flagSetId selector + fsEtag, _, ok := tr.Version(mustSelector(t, "flagSetId=fs1")) + require.True(t, ok) + assert.NotEmpty(t, fsEtag) + + // source selector is tracked for ETag lookups + _, _, ok = tr.Version(mustSelector(t, "source=src1")) + assert.True(t, ok) + + // unknown source -> not tracked + _, _, ok = tr.Version(mustSelector(t, "source=missing")) + assert.False(t, ok) + + // unknown flagSetId -> not tracked + _, _, ok = tr.Version(mustSelector(t, "flagSetId=nope")) + assert.False(t, ok) +} diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go b/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go new file mode 100644 index 000000000..59e83fb00 --- /dev/null +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go @@ -0,0 +1,157 @@ +package ofrep + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gorilla/mux" + "github.com/open-feature/flagd/core/pkg/evaluator" + mock "github.com/open-feature/flagd/core/pkg/evaluator/mock" + "github.com/open-feature/flagd/core/pkg/logger" + "github.com/open-feature/flagd/core/pkg/model" + "github.com/open-feature/flagd/core/pkg/service/ofrep" + "github.com/open-feature/flagd/core/pkg/store" + svc "github.com/open-feature/flagd/flagd/pkg/service" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" +) + +type fakeVersioner struct { + etag string + lastModified int64 + ok bool +} + +func (f fakeVersioner) Version(_ store.Selector) (string, int64, bool) { + return f.etag, f.lastModified, f.ok +} + +func newBulkRequest(t *testing.T, selectorHeader, clientEtag string) *http.Request { + t.Helper() + req, err := http.NewRequest(http.MethodPost, "/ofrep/v1/evaluate/flags", bytes.NewReader([]byte{})) + require.NoError(t, err) + req.Host = "flagd.example:8016" + if selectorHeader != "" { + req.Header.Set(svc.FLAGD_SELECTOR_HEADER, selectorHeader) + } + if clientEtag != "" { + req.Header.Set("If-None-Match", clientEtag) + } + return req +} + +func serveBulk(h handler, req *http.Request) *httptest.ResponseRecorder { + recorder := httptest.NewRecorder() + router := mux.NewRouter() + router.HandleFunc(bulkEvaluation, h.HandleBulkEvaluation) + router.ServeHTTP(recorder, req) + return recorder +} + +func TestHandleBulkEvaluation_NotModified(t *testing.T) { + log := logger.NewLogger(nil, false) + eval := mock.NewMockIEvaluator(gomock.NewController(t)) + // evaluation must be skipped on a 304 + eval.EXPECT().ResolveAllValues(gomock.Any(), gomock.Any(), gomock.Any()).Times(0) + + h := handler{ + Logger: log, + evaluator: eval, + versioner: fakeVersioner{etag: "abc123", ok: true}, + sseEnabled: true, + sseInactivityDelaySec: 120, + } + + recorder := serveBulk(h, newBulkRequest(t, "flagSetId=fs1", `"abc123"`)) + + assert.Equal(t, http.StatusNotModified, recorder.Code) + assert.Equal(t, `"abc123"`, recorder.Header().Get("ETag")) +} + +func TestHandleBulkEvaluation_AdvertisesEventStreams(t *testing.T) { + log := logger.NewLogger(nil, false) + + tests := []struct { + name string + publicURL string + selectorHeader string + wantOrigin string // expected endpoint.origin ("" => omitted) + wantRequestURI string + }{ + { + name: "flagSetId selector, origin omitted", + selectorHeader: "flagSetId=fs1", + wantOrigin: "", + wantRequestURI: "/ofrep/v1/sse?channels=fs1", + }, + { + name: "no selector (catch-all), origin omitted", + selectorHeader: "", + wantOrigin: "", + wantRequestURI: "/ofrep/v1/sse", + }, + { + name: "public url sets origin", + publicURL: "https://flags.example.com/", + selectorHeader: "flagSetId=fs1", + wantOrigin: "https://flags.example.com", + wantRequestURI: "/ofrep/v1/sse?channels=fs1", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + eval := mock.NewMockIEvaluator(gomock.NewController(t)) + eval.EXPECT().ResolveAllValues(gomock.Any(), gomock.Any(), gomock.Any()). + Return([]evaluator.AnyValue{}, model.Metadata{}, nil) + + h := handler{ + Logger: log, + evaluator: eval, + versioner: fakeVersioner{etag: "etag-1", ok: true}, + sseEnabled: true, + sseInactivityDelaySec: 120, + ssePublicURL: tt.publicURL, + } + + recorder := serveBulk(h, newBulkRequest(t, tt.selectorHeader, "")) + + require.Equal(t, http.StatusOK, recorder.Code) + assert.Equal(t, `"etag-1"`, recorder.Header().Get("ETag")) + + var resp ofrep.BulkEvaluationResponse + require.NoError(t, json.Unmarshal(recorder.Body.Bytes(), &resp)) + require.Len(t, resp.EventStreams, 1) + + es := resp.EventStreams[0] + assert.Equal(t, "sse", es.Type) + assert.Equal(t, 120, es.InactivityDelaySec) + require.NotNil(t, es.Endpoint) + assert.Equal(t, tt.wantOrigin, es.Endpoint.Origin) + assert.Equal(t, tt.wantRequestURI, es.Endpoint.RequestUri) + }) + } +} + +func TestHandleBulkEvaluation_SSEDisabled_NoEventStreams(t *testing.T) { + log := logger.NewLogger(nil, false) + eval := mock.NewMockIEvaluator(gomock.NewController(t)) + eval.EXPECT().ResolveAllValues(gomock.Any(), gomock.Any(), gomock.Any()). + Return([]evaluator.AnyValue{}, model.Metadata{}, nil) + + // sseEnabled false, no versioner: behaves like the legacy handler + h := handler{Logger: log, evaluator: eval} + + recorder := serveBulk(h, newBulkRequest(t, "", "")) + + require.Equal(t, http.StatusOK, recorder.Code) + assert.Empty(t, recorder.Header().Get("ETag")) + + var resp ofrep.BulkEvaluationResponse + require.NoError(t, json.Unmarshal(recorder.Body.Bytes(), &resp)) + assert.Empty(t, resp.EventStreams) +} diff --git a/test/zero-downtime-flagd-proxy/go.sum b/test/zero-downtime-flagd-proxy/go.sum index 4320e232b..a33083278 100644 --- a/test/zero-downtime-flagd-proxy/go.sum +++ b/test/zero-downtime-flagd-proxy/go.sum @@ -19,13 +19,10 @@ go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbE go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= -go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= -go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= -go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= @@ -33,10 +30,9 @@ golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= -gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 h1:RmoJA1ujG+/lRGNfUnOMfhCy5EipVMyvUE+KNbPbTlw= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= -google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE= -google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 h1:m8qni9SQFH0tJc1X0vmnpw/0t+AImlSvp30sEupozUg= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.80.0 h1:Xr6m2WmWZLETvUNvIUmeD5OAagMw3FiKmMlTdViWsHM= +google.golang.org/grpc v1.80.0/go.mod h1:ho/dLnxwi3EDJA4Zghp7k2Ec1+c2jqup0bFkw07bwF4= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= From a1bf5080b865bd54ff1aa5cac2ff8e6cafcf6e7e Mon Sep 17 00:00:00 2001 From: Jamie Sinn Date: Tue, 4 Aug 2026 16:58:42 -0400 Subject: [PATCH 2/9] chore: go mod tidy Signed-off-by: Jamie Sinn --- test/zero-downtime-flagd-proxy/go.sum | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/test/zero-downtime-flagd-proxy/go.sum b/test/zero-downtime-flagd-proxy/go.sum index a33083278..4320e232b 100644 --- a/test/zero-downtime-flagd-proxy/go.sum +++ b/test/zero-downtime-flagd-proxy/go.sum @@ -19,10 +19,13 @@ go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbE go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= +go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= +go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= +go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= @@ -30,9 +33,10 @@ golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 h1:m8qni9SQFH0tJc1X0vmnpw/0t+AImlSvp30sEupozUg= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= -google.golang.org/grpc v1.80.0 h1:Xr6m2WmWZLETvUNvIUmeD5OAagMw3FiKmMlTdViWsHM= -google.golang.org/grpc v1.80.0/go.mod h1:ho/dLnxwi3EDJA4Zghp7k2Ec1+c2jqup0bFkw07bwF4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 h1:RmoJA1ujG+/lRGNfUnOMfhCy5EipVMyvUE+KNbPbTlw= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE= +google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= From 46b79c72b287927fd745609a5afdb8546f920261 Mon Sep 17 00:00:00 2001 From: Jamie Sinn Date: Tue, 4 Aug 2026 17:04:49 -0400 Subject: [PATCH 3/9] docs: update docs Signed-off-by: Jamie Sinn --- docs/reference/flagd-cli/flagd_start.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/docs/reference/flagd-cli/flagd_start.md b/docs/reference/flagd-cli/flagd_start.md index 392cd24d9..a36538875 100644 --- a/docs/reference/flagd-cli/flagd_start.md +++ b/docs/reference/flagd-cli/flagd_start.md @@ -24,6 +24,9 @@ flagd start [flags] -R, --max-request-header int Maximum allowed request header size in bytes. Requests exceeding this are rejected with HTTP 431. Set to 0 to use Go's built-in default (1 MiB). WARNING: setting a very large or zero value may allow memory exhaustion from oversized headers. (default 1000000) -t, --metrics-exporter string Set the metrics exporter. Default(if unset) is Prometheus. Can be override to otel - OpenTelemetry metric exporter. Overriding to otel require otelCollectorURI to be present -r, --ofrep-port int32 ofrep service port (default 8016) + --ofrep-sse-enabled Enable the OFREP SSE change-notification endpoint (ADR-0008) at /ofrep/v1/sse on the ofrep port. Defaults to true. (default true) + --ofrep-sse-inactivity-delay int Inactivity delay (seconds) advertised to OFREP SSE clients in the eventStreams block. Clients close idle connections after this. Defaults to 120. (default 120) + --ofrep-sse-public-url string Origin (scheme://host) advertised as the OFREP SSE eventStreams endpoint.origin. Omitted when empty, so clients resolve the requestUri against the OFREP base URL. Set when flagd is behind a proxy. -A, --otel-ca-path string tls certificate authority path to use with OpenTelemetry collector -D, --otel-cert-path string tls certificate path to use with OpenTelemetry collector -o, --otel-collector-uri string Set the grpc URI of the OpenTelemetry collector for flagd runtime. If unset, the collector setup will be ignored and traces will not be exported. From ca5dc0ef511f3b3b3e2df94967f735ad54842793 Mon Sep 17 00:00:00 2001 From: Jamie Sinn Date: Tue, 4 Aug 2026 17:11:56 -0400 Subject: [PATCH 4/9] chore: fix complexity flag from sonar. Signed-off-by: Jamie Sinn --- .../service/flag-evaluation/ofrep/handler.go | 64 ++++++++++++------- 1 file changed, 41 insertions(+), 23 deletions(-) diff --git a/flagd/pkg/service/flag-evaluation/ofrep/handler.go b/flagd/pkg/service/flag-evaluation/ofrep/handler.go index 484285d21..c10cef0f8 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/handler.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/handler.go @@ -171,18 +171,12 @@ func (h *handler) HandleBulkEvaluation(w http.ResponseWriter, r *http.Request) { } ctx := context.WithValue(r.Context(), store.SelectorContextKey{}, selector) - // Conditional evaluation (ADR-0008): if the client already holds the current config - // version, short-circuit with 304 Not Modified instead of re-serving the flags. - etag, lastModified, hasVersion := "", int64(0), false - if h.versioner != nil { - etag, lastModified, hasVersion = h.versioner.Version(selector) - } - if hasVersion && etag != "" { - w.Header().Set("ETag", quoteETag(etag)) - if clientEtag := requestETag(r); clientEtag != "" && normalizeETag(clientEtag) == etag { - w.WriteHeader(http.StatusNotModified) - return - } + // Conditional evaluation (ADR-0008): short-circuit with 304 when the client already holds + // the current config version. + lastModified, notModified := h.applyConditionalETag(w, r, selector) + if notModified { + w.WriteHeader(http.StatusNotModified) + return } evaluations, metadata, err := h.evaluator.ResolveAllValues(ctx, requestID, evaluationContext) @@ -197,19 +191,43 @@ func (h *handler) HandleBulkEvaluation(w http.ResponseWriter, r *http.Request) { res := ofrep.BulkEvaluationContextErrorFrom(model.GeneralErrorCode, fmt.Sprintf("Bulk evaluation failed. Tracking ID: %s", requestID)) h.writeJSONToResponse(http.StatusInternalServerError, res, w) - } else { - response := ofrep.BulkEvaluationResponseFrom(evaluations, metadata) - if h.sseEnabled { - response.EventStreams = h.eventStreams(selector) - if lastModified > 0 { - if response.Metadata == nil { - response.Metadata = model.Metadata{} - } - response.Metadata["flagConfigLastModified"] = lastModified - } + return + } + + h.writeJSONToResponse(http.StatusOK, h.bulkResponse(selector, evaluations, metadata, lastModified), w) +} + +// applyConditionalETag resolves the current config version for the selector, sets the ETag +// response header, and reports the lastModified time plus whether the request can be answered +// with 304 Not Modified. It is a no-op (returns notModified=false) when SSE/versioning is off. +func (h *handler) applyConditionalETag(w http.ResponseWriter, r *http.Request, selector store.Selector) (lastModified int64, notModified bool) { + if h.versioner == nil { + return 0, false + } + etag, lastModified, ok := h.versioner.Version(selector) + if !ok || etag == "" { + return lastModified, false + } + w.Header().Set("ETag", quoteETag(etag)) + clientEtag := requestETag(r) + return lastModified, clientEtag != "" && normalizeETag(clientEtag) == etag +} + +// bulkResponse assembles the OFREP bulk response, adding the ADR-0008 eventStreams block and +// lastModified metadata when SSE is enabled. +func (h *handler) bulkResponse(selector store.Selector, evaluations []evaluator.AnyValue, metadata model.Metadata, lastModified int64) ofrep.BulkEvaluationResponse { + response := ofrep.BulkEvaluationResponseFrom(evaluations, metadata) + if !h.sseEnabled { + return response + } + response.EventStreams = h.eventStreams(selector) + if lastModified > 0 { + if response.Metadata == nil { + response.Metadata = model.Metadata{} } - h.writeJSONToResponse(http.StatusOK, response, w) + response.Metadata["flagConfigLastModified"] = lastModified } + return response } // eventStreams builds the ADR-0008 eventStreams advertisement pointing OFREP clients back at From 4ce5e09268c440f399c7c52a5272ecf8a5f231cd Mon Sep 17 00:00:00 2001 From: Jamie Sinn Date: Tue, 4 Aug 2026 17:32:39 -0400 Subject: [PATCH 5/9] chore: address PR comments from coderabbit Signed-off-by: Jamie Sinn --- .../service/flag-evaluation/ofrep/handler.go | 19 ++--- .../flag-evaluation/ofrep/ofrep_service.go | 12 +++- .../flag-evaluation/ofrep/sse/tracker.go | 23 ++++++ .../flag-evaluation/ofrep/sse/tracker_test.go | 49 +++++++++++++ .../flag-evaluation/ofrep/sse_bulk_test.go | 71 +++++++++++++++++++ 5 files changed, 164 insertions(+), 10 deletions(-) diff --git a/flagd/pkg/service/flag-evaluation/ofrep/handler.go b/flagd/pkg/service/flag-evaluation/ofrep/handler.go index c10cef0f8..35abca85f 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/handler.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/handler.go @@ -209,8 +209,13 @@ func (h *handler) applyConditionalETag(w http.ResponseWriter, r *http.Request, s return lastModified, false } w.Header().Set("ETag", quoteETag(etag)) - clientEtag := requestETag(r) - return lastModified, clientEtag != "" && normalizeETag(clientEtag) == etag + + if trigger := r.URL.Query().Get(flagConfigEtagParam); trigger != "" { + h.Logger.Debug(fmt.Sprintf("bulk refetch triggered by %s=%s", flagConfigEtagParam, trigger)) + } + + clientCacheETag := r.Header.Get("If-None-Match") + return lastModified, clientCacheETag != "" && normalizeETag(clientCacheETag) == etag } // bulkResponse assembles the OFREP bulk response, adding the ADR-0008 eventStreams block and @@ -257,12 +262,10 @@ func (h *handler) eventStreams(selector store.Selector) []ofrep.EventStream { }} } -func requestETag(r *http.Request) string { - if e := r.URL.Query().Get("flagConfigEtag"); e != "" { - return e - } - return r.Header.Get("If-None-Match") -} +// flagConfigEtagParam is the ADR-0008 query parameter carrying the config version that an SSE +// refetch event advertised. It is trigger metadata only; the conditional response is decided by +// the If-None-Match header. See applyConditionalETag. +const flagConfigEtagParam = "flagConfigEtag" // normalizeETag strips optional surrounding quotes so quoted and unquoted forms compare equal. func normalizeETag(etag string) string { diff --git a/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service.go b/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service.go index 3ce9064ad..ededd7fec 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service.go @@ -50,13 +50,21 @@ func NewOfrepService( ) (*Service, error) { corsMiddleware := corsmw.New(origins) + // SSE requires a flag store to watch; without one we cannot serve or advertise it, so treat + // a missing store as SSE disabled rather than crashing later in the tracker goroutine. + sseEnabled := cfg.SSEEnabled + if sseEnabled && flagStore == nil { + cfg.Logger.Warn("OFREP SSE requested but no flag store was provided; disabling SSE") + sseEnabled = false + } + var sseService *sse.Service sseCfg := SSEConfig{ - Enabled: cfg.SSEEnabled, + Enabled: sseEnabled, InactivityDelaySec: cfg.SSEInactivityDelaySec, PublicURL: cfg.SSEPublicURL, } - if cfg.SSEEnabled { + if sseEnabled { sseService = sse.New(flagStore, sse.Config{Logger: cfg.Logger}) sseCfg.Versioner = sseService.Tracker() } diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go index a0bd21889..3a0f291cc 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go @@ -67,6 +67,13 @@ func NewTracker(log *logger.Logger, s store.IStore, es *eventsource.Server) *Tra // Run subscribes to all flag changes and publishes refetch events until ctx is cancelled. // It blocks, so it is intended to run in its own goroutine. func (t *Tracker) Run(ctx context.Context) { + if t.store == nil { + if t.logger != nil { + t.logger.Warn("ofrep sse tracker has no flag store; change notifications disabled") + } + return + } + watcher := make(chan store.FlagQueryResult, 1) t.store.Watch(ctx, &store.Selector{}, watcher) @@ -82,6 +89,22 @@ func (t *Tracker) Run(ctx context.Context) { t.publish(ch) } } + + // The store closed the watcher. That happens on context cancellation (expected shutdown) or + // on a selector/iterator error inside the store (unexpected). Distinguish them: on the + // unexpected path publishing has stopped and t.versions is now frozen, which would make the + // bulk handler keep returning stale 304s forever, so surface it and invalidate the versions + // (Version then reports no ETag and clients are served fresh flags instead of a stale 304). + if ctx.Err() != nil { + return + } + if t.logger != nil { + t.logger.Error("ofrep sse tracker stopped watching the flag store unexpectedly; " + + "refetch events will no longer be published and cached ETags are invalidated") + } + t.mu.Lock() + t.versions = map[string]version{} + t.mu.Unlock() } // Version returns the current config ETag and last-modified time (unix seconds) for the diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go index 580397767..7883734be 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go @@ -1,8 +1,10 @@ package sse import ( + "context" "testing" + "github.com/open-feature/flagd/core/pkg/logger" "github.com/open-feature/flagd/core/pkg/model" "github.com/open-feature/flagd/core/pkg/store" "github.com/stretchr/testify/assert" @@ -83,6 +85,53 @@ func TestTracker_Update_NilFlagSetIdNotSubscribable(t *testing.T) { assert.NotContains(t, changed, store.NilFlagSetId(), "internal nilFlagSetId must not be a subscribable channel") } +func TestTracker_Run_NilStoreDoesNotPanic(t *testing.T) { + tr := NewTracker(logger.NewLogger(nil, false), nil, nil) + // Run must return promptly instead of dereferencing the nil store (which would panic in + // the background goroutine and terminate the process). + require.NotPanics(t, func() { tr.Run(context.Background()) }) +} + +// closingStore closes the watcher immediately without emitting, simulating store.Watch's +// error-close path. +type closingStore struct{} + +func (closingStore) Get(context.Context, string, *store.Selector) (model.Flag, model.Metadata, error) { + return model.Flag{}, nil, nil +} +func (closingStore) GetAll(context.Context, *store.Selector) ([]model.Flag, model.Metadata, error) { + return nil, nil, nil +} +func (closingStore) Watch(_ context.Context, _ *store.Selector, watcher chan<- store.FlagQueryResult) { + close(watcher) +} +func (closingStore) Update(string, []model.Flag, model.Metadata, bool) {} + +func TestTracker_Run_UnexpectedCloseInvalidatesVersions(t *testing.T) { + tr := NewTracker(logger.NewLogger(nil, false), closingStore{}, nil) + tr.versions = map[string]version{allKey: {etag: "frozen"}} + + // context is NOT cancelled -> the watcher closing is unexpected (store error path) + tr.Run(context.Background()) + + _, _, ok := tr.Version(store.Selector{}) + assert.False(t, ok, "frozen versions must be invalidated so the bulk handler stops serving stale 304s") +} + +func TestTracker_Run_ContextCancelKeepsVersions(t *testing.T) { + tr := NewTracker(logger.NewLogger(nil, false), closingStore{}, nil) + tr.versions = map[string]version{allKey: {etag: "current"}} + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // expected shutdown + + tr.Run(ctx) + + etag, _, ok := tr.Version(store.Selector{}) + assert.True(t, ok, "versions must be retained on a normal context-cancel shutdown") + assert.Equal(t, "current", etag) +} + func TestTracker_Version(t *testing.T) { tr := &Tracker{versions: map[string]version{}} tr.update([]model.Flag{testFlag("fs1", "a", "on")}) diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go b/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go index 59e83fb00..a4d1c48ad 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go @@ -72,6 +72,77 @@ func TestHandleBulkEvaluation_NotModified(t *testing.T) { assert.Equal(t, `"abc123"`, recorder.Header().Get("ETag")) } +// The 304 decision must use If-None-Match (the client's cached version), not the ADR-0008 +// flagConfigEtag change-trigger metadata. +func TestHandleBulkEvaluation_ConditionalUsesIfNoneMatch(t *testing.T) { + log := logger.NewLogger(nil, false) + const current = "etag-v2" + + tests := []struct { + name string + flagConfigEtag string // query param (change-trigger metadata) + ifNoneMatch string // header (cache validator) + wantStatus int + wantEval bool + }{ + { + name: "stale cache echoing new trigger etag is served fresh (regression)", + flagConfigEtag: "etag-v2", + ifNoneMatch: `"etag-v1"`, + wantStatus: http.StatusOK, + wantEval: true, + }, + { + name: "current cache validator yields 304", + ifNoneMatch: `"etag-v2"`, + wantStatus: http.StatusNotModified, + wantEval: false, + }, + { + name: "trigger etag alone (no validator) is served fresh", + flagConfigEtag: "etag-v2", + wantStatus: http.StatusOK, + wantEval: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + eval := mock.NewMockIEvaluator(gomock.NewController(t)) + expect := eval.EXPECT().ResolveAllValues(gomock.Any(), gomock.Any(), gomock.Any()). + Return([]evaluator.AnyValue{}, model.Metadata{}, nil) + if tt.wantEval { + expect.Times(1) + } else { + expect.Times(0) + } + + h := handler{ + Logger: log, + evaluator: eval, + versioner: fakeVersioner{etag: current, ok: true}, + sseEnabled: true, + } + + url := "/ofrep/v1/evaluate/flags" + if tt.flagConfigEtag != "" { + url += "?flagConfigEtag=" + tt.flagConfigEtag + } + req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader([]byte{})) + require.NoError(t, err) + req.Header.Set(svc.FLAGD_SELECTOR_HEADER, "flagSetId=fs1") + if tt.ifNoneMatch != "" { + req.Header.Set("If-None-Match", tt.ifNoneMatch) + } + + recorder := serveBulk(h, req) + + assert.Equal(t, tt.wantStatus, recorder.Code) + assert.Equal(t, `"`+current+`"`, recorder.Header().Get("ETag")) + }) + } +} + func TestHandleBulkEvaluation_AdvertisesEventStreams(t *testing.T) { log := logger.NewLogger(nil, false) From 63019514dd59a9cc502f3d8c5f71922527145bf4 Mon Sep 17 00:00:00 2001 From: Jamie Sinn Date: Wed, 19 Aug 2026 15:39:52 -0400 Subject: [PATCH 6/9] feat: Update SSE to use selector paths instead of raw flagset references. Also use the built in subscriber to clean up the tracking massively. Signed-off-by: Jamie Sinn --- .../service/flag-evaluation/ofrep/handler.go | 38 ++- .../flag-evaluation/ofrep/handler_test.go | 4 +- .../flag-evaluation/ofrep/sse/handler.go | 71 ++--- .../flag-evaluation/ofrep/sse/service.go | 26 +- .../flag-evaluation/ofrep/sse/service_test.go | 76 ++++- .../flag-evaluation/ofrep/sse/tracker.go | 288 ++++++++---------- .../flag-evaluation/ofrep/sse/tracker_test.go | 237 +++++++++----- .../flag-evaluation/ofrep/sse_bulk_test.go | 56 +++- 8 files changed, 470 insertions(+), 326 deletions(-) diff --git a/flagd/pkg/service/flag-evaluation/ofrep/handler.go b/flagd/pkg/service/flag-evaluation/ofrep/handler.go index 35abca85f..ae3614b70 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/handler.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/handler.go @@ -18,6 +18,7 @@ import ( "github.com/open-feature/flagd/core/pkg/telemetry" "github.com/open-feature/flagd/flagd/pkg/service" evalservice "github.com/open-feature/flagd/flagd/pkg/service/flag-evaluation" + "github.com/open-feature/flagd/flagd/pkg/service/flag-evaluation/ofrep/sse" metricsmw "github.com/open-feature/flagd/flagd/pkg/service/middleware/metrics" "github.com/rs/xid" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" @@ -31,11 +32,12 @@ const ( bulkEvaluation = "/ofrep/v1/evaluate/{path:flags\\/|flags}" ) -// configVersioner resolves the current config ETag / last-modified time for a selector so the -// bulk handler can serve conditional (ETag/304) responses consistent with the SSE stream. -// Implemented by the OFREP SSE change tracker; nil when SSE is disabled. +// configVersioner resolves the current config ETag / last-modified time for an SSE channel so +// the bulk handler can serve conditional (ETag/304) responses consistent with the SSE stream. +// Implemented by the OFREP SSE change tracker; nil when SSE is disabled. ok is false whenever no +// stream for the channel is live, which is the normal state before a client connects. type configVersioner interface { - Version(selector store.Selector) (etag string, lastModified int64, ok bool) + Version(channel string) (etag string, lastModified int64, ok bool) } type handler struct { @@ -173,7 +175,7 @@ func (h *handler) HandleBulkEvaluation(w http.ResponseWriter, r *http.Request) { // Conditional evaluation (ADR-0008): short-circuit with 304 when the client already holds // the current config version. - lastModified, notModified := h.applyConditionalETag(w, r, selector) + lastModified, notModified := h.applyConditionalETag(w, r, selectorExpression) if notModified { w.WriteHeader(http.StatusNotModified) return @@ -194,17 +196,18 @@ func (h *handler) HandleBulkEvaluation(w http.ResponseWriter, r *http.Request) { return } - h.writeJSONToResponse(http.StatusOK, h.bulkResponse(selector, evaluations, metadata, lastModified), w) + h.writeJSONToResponse(http.StatusOK, + h.bulkResponse(selectorExpression, evaluations, metadata, lastModified), w) } // applyConditionalETag resolves the current config version for the selector, sets the ETag // response header, and reports the lastModified time plus whether the request can be answered // with 304 Not Modified. It is a no-op (returns notModified=false) when SSE/versioning is off. -func (h *handler) applyConditionalETag(w http.ResponseWriter, r *http.Request, selector store.Selector) (lastModified int64, notModified bool) { +func (h *handler) applyConditionalETag(w http.ResponseWriter, r *http.Request, channel string) (lastModified int64, notModified bool) { if h.versioner == nil { return 0, false } - etag, lastModified, ok := h.versioner.Version(selector) + etag, lastModified, ok := h.versioner.Version(channel) if !ok || etag == "" { return lastModified, false } @@ -220,12 +223,12 @@ func (h *handler) applyConditionalETag(w http.ResponseWriter, r *http.Request, s // bulkResponse assembles the OFREP bulk response, adding the ADR-0008 eventStreams block and // lastModified metadata when SSE is enabled. -func (h *handler) bulkResponse(selector store.Selector, evaluations []evaluator.AnyValue, metadata model.Metadata, lastModified int64) ofrep.BulkEvaluationResponse { +func (h *handler) bulkResponse(selectorExpression string, evaluations []evaluator.AnyValue, metadata model.Metadata, lastModified int64) ofrep.BulkEvaluationResponse { response := ofrep.BulkEvaluationResponseFrom(evaluations, metadata) if !h.sseEnabled { return response } - response.EventStreams = h.eventStreams(selector) + response.EventStreams = h.eventStreams(selectorExpression) if lastModified > 0 { if response.Metadata == nil { response.Metadata = model.Metadata{} @@ -236,20 +239,15 @@ func (h *handler) bulkResponse(selector store.Selector, evaluations []evaluator. } // eventStreams builds the ADR-0008 eventStreams advertisement pointing OFREP clients back at -// this flagd's SSE endpoint. The channel is the request's flagSetId; the internal nilFlagSetId -// and a missing flagSetId both map to the catch-all channel (no `channels` parameter). +// this flagd's SSE endpoint. The advertised channel is the request's own selector expression, so +// the stream covers exactly the flags the client just evaluated. // // It uses the structured `endpoint` form and omits origin unless a public URL is configured, so // the client resolves the requestUri against the OFREP base URL it is already talking to. -func (h *handler) eventStreams(selector store.Selector) []ofrep.EventStream { - channel := selector.FlagSetId() - if channel == store.NilFlagSetId() { - channel = "" - } - +func (h *handler) eventStreams(selectorExpression string) []ofrep.EventStream { requestURI := ssePath - if channel != "" { - requestURI += "?channels=" + url.QueryEscape(channel) + if selectorExpression != "" { + requestURI += "?" + sse.ChannelParam + "=" + url.QueryEscape(selectorExpression) } return []ofrep.EventStream{{ diff --git a/flagd/pkg/service/flag-evaluation/ofrep/handler_test.go b/flagd/pkg/service/flag-evaluation/ofrep/handler_test.go index fdb01784b..a530b03cb 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/handler_test.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/handler_test.go @@ -472,7 +472,7 @@ func TestHandlerRecordsSingleEvaluationMetrics(t *testing.T) { eval.EXPECT(). ResolveAsAnyValue(gomock.Any(), gomock.Any(), flagKey, gomock.Any()). Return(test.evaluation) - handler := NewOfrepHandler(logger.NewLogger(nil, false), eval, nil, nil, metrics, "flagd") + handler := NewOfrepHandler(logger.NewLogger(nil, false), eval, nil, nil, metrics, "flagd", SSEConfig{}) request := httptest.NewRequest(http.MethodPost, "/ofrep/v1/evaluate/flags/"+flagKey, nil) response := httptest.NewRecorder() @@ -490,7 +490,7 @@ func TestHandlerRecordsEachBulkEvaluationMetric(t *testing.T) { eval := mock.NewMockIEvaluator(gomock.NewController(t)) eval.EXPECT().ResolveAllValues(gomock.Any(), gomock.Any(), gomock.Any()). Return(evaluations, model.Metadata{}, nil) - handler := NewOfrepHandler(logger.NewLogger(nil, false), eval, nil, nil, metrics, "flagd") + handler := NewOfrepHandler(logger.NewLogger(nil, false), eval, nil, nil, metrics, "flagd", SSEConfig{}) request := httptest.NewRequest(http.MethodPost, "/ofrep/v1/evaluate/flags", nil) response := httptest.NewRecorder() diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/handler.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/handler.go index 44c51e97c..b70c113eb 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse/handler.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/handler.go @@ -2,49 +2,34 @@ package sse import ( "net/http" - "sync" -) - -// channelParam carries the ADR-0008 channel token (the flagSetId); empty selects the catch-all. -const channelParam = "channels" - -func channelFromRequest(r *http.Request) string { - return r.URL.Query().Get(channelParam) -} - -// activeChannels is a reference-counted set of channels with at least one live subscriber, so -// the heartbeat loop knows which channels to ping (the eventsource server exposes no registry). -type activeChannels struct { - mu sync.Mutex - counts map[string]int -} -func newActiveChannels() *activeChannels { - return &activeChannels{counts: map[string]int{}} -} - -func (a *activeChannels) add(channel string) { - a.mu.Lock() - defer a.mu.Unlock() - a.counts[channel]++ -} - -func (a *activeChannels) remove(channel string) { - a.mu.Lock() - defer a.mu.Unlock() - if a.counts[channel] <= 1 { - delete(a.counts, channel) - return - } - a.counts[channel]-- -} + "github.com/open-feature/flagd/core/pkg/store" +) -func (a *activeChannels) snapshot() []string { - a.mu.Lock() - defer a.mu.Unlock() - channels := make([]string, 0, len(a.counts)) - for ch := range a.counts { - channels = append(channels, ch) - } - return channels +// ChannelParam carries the selector expression the client wants change notifications for, using +// the same syntax as the Flagd-Selector header; empty selects every flag. It is exported so the +// bulk handler advertises the same parameter this handler reads. +const ChannelParam = "channels" + +// Handler resolves the request's selector, takes a reference on the matching subscription so the +// store watch stays alive, and streams events until the client disconnects. +func (svc *Service) Handler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + channel := r.URL.Query().Get(ChannelParam) + selector, err := store.NewSelector(channel) + if err != nil { + // not echoing the expression back: it is unescaped client input + http.Error(w, "invalid selector in the 'channels' parameter", http.StatusBadRequest) + return + } + + release, err := svc.tracker.Subscribe(channel, selector) + if err != nil { + http.Error(w, "unable to subscribe to flag changes", http.StatusServiceUnavailable) + return + } + defer release() + + svc.es.Handler(channel).ServeHTTP(w, r) + }) } diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/service.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/service.go index 97109b7b0..2e769e408 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse/service.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/service.go @@ -2,7 +2,6 @@ package sse import ( "context" - "net/http" "time" "github.com/launchdarkly/eventsource" @@ -25,7 +24,6 @@ type Service struct { logger *logger.Logger es *eventsource.Server tracker *Tracker - active *activeChannels heartbeatInterval time.Duration } @@ -45,7 +43,6 @@ func New(s store.IStore, cfg Config) *Service { logger: cfg.Logger, es: es, tracker: NewTracker(cfg.Logger, s, es), - active: newActiveChannels(), heartbeatInterval: heartbeat, } } @@ -54,36 +51,25 @@ func New(s store.IStore, cfg Config) *Service { // for conditional (ETag/304) evaluation. func (svc *Service) Tracker() *Tracker { return svc.tracker } -// Handler registers the request's channel in the active set (for heartbeats) and delegates to -// the eventsource server, which streams until the client disconnects. -func (svc *Service) Handler() http.Handler { - return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - channel := channelFromRequest(r) - svc.active.add(channel) - defer svc.active.remove(channel) - svc.es.Handler(channel).ServeHTTP(w, r) - }) -} - -// Start runs the change tracker and heartbeat loop until ctx is cancelled, then shuts the -// eventsource server down. It blocks and is intended to run in its own goroutine. +// Start runs the heartbeat loop until ctx is cancelled, then shuts down. It blocks, so it is +// intended to run in its own goroutine. func (svc *Service) Start(ctx context.Context) error { - go svc.tracker.Run(ctx) - ticker := time.NewTicker(svc.heartbeatInterval) defer ticker.Stop() for { select { case <-ctx.Done(): + // Order matters: the tracker's watch goroutines are the only publishers, and + // eventsource.Server.Publish blocks forever once the server is closed. + svc.tracker.Close() svc.es.Close() if svc.logger != nil { svc.logger.Info("shutting down ofrep sse service") } return nil case <-ticker.C: - channels := svc.active.snapshot() - if len(channels) > 0 { + if channels := svc.tracker.Channels(); len(channels) > 0 { svc.es.PublishComment(channels, "keep-alive") } } diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/service_test.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/service_test.go index be62a9b9f..d8365ae20 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse/service_test.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/service_test.go @@ -2,7 +2,9 @@ package sse import ( "context" + "net/http" "net/http/httptest" + "net/url" "testing" "time" @@ -14,8 +16,8 @@ import ( "github.com/stretchr/testify/require" ) -// TestService_PublishesRefetchOnChange verifies the end-to-end wiring: a client subscribed to a -// flagSetId channel receives an ADR-0008 refetchEvaluation event when that flag set changes. +// TestService_PublishesRefetchOnChange verifies the end-to-end wiring: a client subscribed with +// a flagSetId selector receives an ADR-0008 refetchEvaluation event when that flag set changes. func TestService_PublishesRefetchOnChange(t *testing.T) { log := logger.NewLogger(nil, false) s, err := store.NewStore(log, []string{"src1"}) @@ -31,14 +33,12 @@ func TestService_PublishesRefetchOnChange(t *testing.T) { ts := httptest.NewServer(svc.Handler()) defer ts.Close() - // allow the tracker's initial (empty) snapshot to be consumed and skipped - time.Sleep(100 * time.Millisecond) - - stream, err := eventsource.Subscribe(ts.URL+"?channels=fs1", "") + // the channel token is a selector expression, same syntax as Flagd-Selector + stream, err := eventsource.Subscribe(ts.URL+"?channels="+url.QueryEscape("flagSetId=fs1"), "") require.NoError(t, err) defer stream.Close() - // allow the subscription to register server-side before publishing + // eventsource writes the 200 before registering, so Subscribe returning is not enough time.Sleep(100 * time.Millisecond) s.Update("src1", []model.Flag{testFlag("fs1", "a", "on")}, model.Metadata{"flagSetId": "fs1"}, false) @@ -51,3 +51,65 @@ func TestService_PublishesRefetchOnChange(t *testing.T) { t.Fatal("timed out waiting for refetch event on channel fs1") } } + +// TestService_SelectorScopedNotifications asserts a client is woken only for changes to the +// flags its selector selects, including a source= selector, which previously fell back to the +// catch-all channel and was woken by every change. +func TestService_SelectorScopedNotifications(t *testing.T) { + log := logger.NewLogger(nil, false) + s, err := store.NewStore(log, []string{"src1", "src2"}) + require.NoError(t, err) + + s.Update("src1", []model.Flag{testFlag("fs1", "a", "on")}, model.Metadata{"flagSetId": "fs1"}, false) + s.Update("src2", []model.Flag{testFlag("fs2", "b", "on")}, model.Metadata{"flagSetId": "fs2"}, false) + + svc := New(s, Config{Logger: log, HeartbeatInterval: time.Hour}) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = svc.Start(ctx) }() + + ts := httptest.NewServer(svc.Handler()) + defer ts.Close() + + bySource, err := eventsource.Subscribe(ts.URL+"?channels="+url.QueryEscape("source=src1"), "") + require.NoError(t, err) + defer bySource.Close() + + byFlagSet, err := eventsource.Subscribe(ts.URL+"?channels="+url.QueryEscape("flagSetId=fs2"), "") + require.NoError(t, err) + defer byFlagSet.Close() + + time.Sleep(100 * time.Millisecond) + + s.Update("src2", []model.Flag{testFlag("fs2", "b", "off")}, model.Metadata{"flagSetId": "fs2"}, false) + + select { + case ev := <-byFlagSet.Events: + assert.Contains(t, ev.Data(), refetchEventType) + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for a refetch event on the flagSetId=fs2 stream") + } + + select { + case ev := <-bySource.Events: + t.Fatalf("source=src1 must not be notified about a src2 change, got %q", ev.Data()) + case <-time.After(500 * time.Millisecond): + } +} + +func TestService_Handler_InvalidSelectorReturns400(t *testing.T) { + log := logger.NewLogger(nil, false) + s, err := store.NewStore(log, []string{"src1"}) + require.NoError(t, err) + + svc := New(s, Config{Logger: log, HeartbeatInterval: time.Hour}) + ts := httptest.NewServer(svc.Handler()) + defer ts.Close() + + resp, err := http.Get(ts.URL + "?channels=" + url.QueryEscape("bogusKey=1")) + require.NoError(t, err) + defer resp.Body.Close() + + assert.Equal(t, http.StatusBadRequest, resp.StatusCode) + assert.Empty(t, svc.Tracker().Channels(), "a rejected request must not create a subscription") +} diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go index 3a0f291cc..2712de549 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go @@ -8,7 +8,6 @@ import ( "fmt" "sort" "strconv" - "strings" "sync" "sync/atomic" "time" @@ -19,219 +18,188 @@ import ( "github.com/open-feature/flagd/core/pkg/store" ) -// allChannel receives a refetch event on any flag change. It is advertised to OFREP clients -// that use no flagSetId selector, mirroring a bulk request with no selector. -const allChannel = "" - -// namespacing prefixes for the internal version map, so a flagSetId and a source with the -// same string value never collide. -const ( - allKey = "all" - fsPrefix = "fs:" - srcPrefix = "src:" -) - -func fsKey(id string) string { return fsPrefix + id } -func srcKey(src string) string { return srcPrefix + src } - type version struct { etag string lastModified int64 } -// Tracker owns a global store.Watch subscription. On every flag-configuration change it -// recomputes a per-channel config fingerprint (used as an ETag), publishes an ADR-0008 -// refetchEvaluation event to each affected channel, and answers version lookups for the OFREP -// bulk handler so conditional (ETag/304) evaluation stays consistent with the SSE stream. +// subscription is a single store.Watch, shared by every client on the same channel. It is +// reachable through Tracker.subs only while refs > 0. +type subscription struct { + selector store.Selector // immutable after construction: the store retains &selector + cancel context.CancelFunc + + // ver is nil until the first snapshot lands, and again if the watch dies. + ver atomic.Pointer[version] + + refs int // guarded by Tracker.mu +} + type Tracker struct { logger *logger.Logger store store.IStore es *eventsource.Server - mu sync.RWMutex - versions map[string]version + mu sync.Mutex + subs map[string]*subscription + closed bool + wg sync.WaitGroup eventID atomic.Int64 } -// NewTracker creates a Tracker. Call Run to begin watching the store. func NewTracker(log *logger.Logger, s store.IStore, es *eventsource.Server) *Tracker { return &Tracker{ - logger: log, - store: s, - es: es, - versions: map[string]version{}, + logger: log, + store: s, + es: es, + subs: map[string]*subscription{}, } } -// Run subscribes to all flag changes and publishes refetch events until ctx is cancelled. -// It blocks, so it is intended to run in its own goroutine. -func (t *Tracker) Run(ctx context.Context) { - if t.store == nil { - if t.logger != nil { - t.logger.Warn("ofrep sse tracker has no flag store; change notifications disabled") - } - return +// Subscribe registers interest in a channel, starting a store watch for its selector if this is +// the first subscriber. The returned release must be called exactly once on disconnect. +func (t *Tracker) Subscribe(channel string, selector store.Selector) (func(), error) { + t.mu.Lock() + defer t.mu.Unlock() + + if t.closed { + return nil, fmt.Errorf("ofrep sse tracker is shutting down") + } + + sub, existing := t.subs[channel] + if !existing { + // Subscriptions are shared, so they outlive any single request and are torn down by + // release or Close rather than by a request context. + ctx, cancel := context.WithCancel(context.Background()) + sub = &subscription{selector: selector, cancel: cancel} + t.subs[channel] = sub + t.wg.Add(1) + + watcher := make(chan store.FlagQueryResult, 1) + t.store.Watch(ctx, &sub.selector, watcher) + go t.watch(ctx, channel, sub, watcher) + } + sub.refs++ + + var once sync.Once + return func() { once.Do(func() { t.release(channel, sub) }) }, nil +} + +// release drops one reference, tearing the subscription down when the last one goes. The map +// delete happens under the same lock as the refcount check and before cancel, so a concurrent +// Subscribe cannot attach to a dying subscription. +func (t *Tracker) release(channel string, sub *subscription) { + t.mu.Lock() + sub.refs-- + last := sub.refs <= 0 + if last && t.subs[channel] == sub { + delete(t.subs, channel) } + t.mu.Unlock() - watcher := make(chan store.FlagQueryResult, 1) - t.store.Watch(ctx, &store.Selector{}, watcher) + if last { + sub.cancel() + } +} + +func (t *Tracker) watch(ctx context.Context, channel string, sub *subscription, watcher <-chan store.FlagQueryResult) { + defer t.wg.Done() first := true + // This loop must run until the channel closes: store.Watch's send does not select on + // ctx.Done(), so abandoning the channel would leak the store's goroutine. for res := range watcher { - channels := t.update(res.Flags) + fp := fingerprint(res.Flags) + if prev := sub.ver.Load(); prev != nil && prev.etag == fp { + // A no-op wakeup: an identical re-sync, or a coarser radix watch channel firing + // for a change outside this selector. + continue + } + sub.ver.Store(&version{etag: fp, lastModified: time.Now().Unix()}) + if first { - // The first emission is the initialization snapshot; only seed fingerprints. + // seed only; the connecting client re-fetches unconditionally anyway (ADR-0008) first = false continue } - for _, ch := range channels { - t.publish(ch) - } + t.publish(channel, sub) } - // The store closed the watcher. That happens on context cancellation (expected shutdown) or - // on a selector/iterator error inside the store (unexpected). Distinguish them: on the - // unexpected path publishing has stopped and t.versions is now frozen, which would make the - // bulk handler keep returning stale 304s forever, so surface it and invalidate the versions - // (Version then reports no ETag and clients are served fresh flags instead of a stale 304). if ctx.Err() != nil { - return + return // our own teardown } + + // The store hit a selector/iterator error, so this channel will never fire again. Stop + // serving its ETag, which would otherwise answer 304 from a frozen fingerprint forever. + sub.ver.Store(nil) if t.logger != nil { - t.logger.Error("ofrep sse tracker stopped watching the flag store unexpectedly; " + - "refetch events will no longer be published and cached ETags are invalidated") + t.logger.Error(fmt.Sprintf( + "ofrep sse watch for channel %q ended unexpectedly; its cached ETag is invalidated", channel)) } - t.mu.Lock() - t.versions = map[string]version{} - t.mu.Unlock() } -// Version returns the current config ETag and last-modified time (unix seconds) for the -// channel matching the given selector. ok is false when no version is tracked for the -// selector (e.g. an unknown source), letting the caller skip conditional handling. -func (t *Tracker) Version(selector store.Selector) (etag string, lastModified int64, ok bool) { - var key string - switch { - case selector.IsEmpty(): - key = allKey - case selector.FlagSetId() != "": - key = fsKey(selector.FlagSetId()) - case selector.Source() != "": - key = srcKey(selector.Source()) - default: - return "", 0, false +// publish must be called only after the new version is stored, so a client that refetches the +// instant it receives the event cannot read the previous ETag and be served a stale 304. +func (t *Tracker) publish(channel string, sub *subscription) { + v := sub.ver.Load() + if v == nil { + return } - t.mu.RLock() - defer t.mu.RUnlock() - v, ok := t.versions[key] - return v.etag, v.lastModified, ok + id := strconv.FormatInt(t.eventID.Add(1), 10) + t.es.Publish([]string{channel}, newRefetchEvent(id, v.etag, v.lastModified)) + if t.logger != nil { + t.logger.Debug(fmt.Sprintf("published refetch event to channel %q (etag=%s)", channel, v.etag)) + } } -// update recomputes fingerprints for the catch-all, per-flagSetId and per-source groups, -// swaps them into the version map (preserving lastModified when a fingerprint is unchanged) -// and returns the eventsource channels that should be notified. -func (t *Tracker) update(flags []model.Flag) []string { - now := time.Now().Unix() - - fsGroups := map[string][]model.Flag{} - srcGroups := map[string][]model.Flag{} - for _, f := range flags { - fsGroups[f.FlagSetId] = append(fsGroups[f.FlagSetId], f) - srcGroups[f.Source] = append(srcGroups[f.Source], f) - } +// Version returns the current config ETag and last-modified time (unix seconds) for the channel +// matching the given selector expression. ok is false when no stream for it is live, so the +// caller skips conditional handling. +func (t *Tracker) Version(channel string) (etag string, lastModified int64, ok bool) { + t.mu.Lock() + sub, exists := t.subs[channel] + t.mu.Unlock() - newVersions := make(map[string]version, len(fsGroups)+len(srcGroups)+1) - newVersions[allKey] = version{etag: fingerprint(flags)} - for id, g := range fsGroups { - newVersions[fsKey(id)] = version{etag: fingerprint(g)} + if !exists { + return "", 0, false } - for src, g := range srcGroups { - newVersions[srcKey(src)] = version{etag: fingerprint(g)} + v := sub.ver.Load() + if v == nil { + return "", 0, false } + return v.etag, v.lastModified, true +} +// Channels returns the channels with at least one live subscriber. +func (t *Tracker) Channels() []string { t.mu.Lock() defer t.mu.Unlock() - // carry lastModified forward when the fingerprint is unchanged - for key, nv := range newVersions { - if old, exists := t.versions[key]; exists && old.etag == nv.etag { - nv.lastModified = old.lastModified - } else { - nv.lastModified = now - } - newVersions[key] = nv - } - - changed := t.changedChannels(newVersions, fsGroups) - t.versions = newVersions - return changed -} - -// changedChannels compares the previous version map (t.versions, still held) with the freshly -// computed one and returns the eventsource channels that clients subscribe to and whose config -// changed: the catch-all channel plus any created/updated/removed flagSetId channels. Source -// channels are tracked for ETag lookups but are not (yet) directly subscribable. -func (t *Tracker) changedChannels(newVersions map[string]version, fsGroups map[string][]model.Flag) []string { - etagChanged := func(key string) bool { - old, oldOK := t.versions[key] - nv, newOK := newVersions[key] - if oldOK != newOK { - return true - } - return old.etag != nv.etag - } - - notify := map[string]struct{}{} - if etagChanged(allKey) { - notify[allChannel] = struct{}{} - } - - nilID := store.NilFlagSetId() - for id := range fsGroups { - if id == nilID { - continue // internal flagSetId, never subscribable - } - if etagChanged(fsKey(id)) { - notify[id] = struct{}{} - } - } - // flagSetId channels that existed before but have no flags now (whole set deleted) - for key := range t.versions { - id, ok := strings.CutPrefix(key, fsPrefix) - if !ok || id == nilID { - continue - } - if _, present := fsGroups[id]; !present { - notify[id] = struct{}{} - } - } - - channels := make([]string, 0, len(notify)) - for ch := range notify { - channels = append(channels, ch) + channels := make([]string, 0, len(t.subs)) + for channel := range t.subs { + channels = append(channels, channel) } return channels } -// publish emits a refetch event to a single eventsource channel using that channel's current -// ETag and lastModified. -func (t *Tracker) publish(channel string) { - key := allKey - if channel != allChannel { - key = fsKey(channel) +// Close stops accepting subscriptions and waits for every watch goroutine to exit. It must +// complete before the eventsource server is closed, since a publish after that blocks forever. +func (t *Tracker) Close() { + t.mu.Lock() + t.closed = true + subs := make([]*subscription, 0, len(t.subs)) + for _, sub := range t.subs { + subs = append(subs, sub) } + t.subs = map[string]*subscription{} + t.mu.Unlock() - t.mu.RLock() - v := t.versions[key] - t.mu.RUnlock() - - id := strconv.FormatInt(t.eventID.Add(1), 10) - t.es.Publish([]string{channel}, newRefetchEvent(id, v.etag, v.lastModified)) - if t.logger != nil { - t.logger.Debug(fmt.Sprintf("published refetch event to channel %q (etag=%s)", channel, v.etag)) + for _, sub := range subs { + sub.cancel() } + t.wg.Wait() } // fingerprint produces a deterministic, restart-stable hash of a group of flag definitions. diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go index 7883734be..d85e4f81b 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go @@ -2,7 +2,11 @@ package sse import ( "context" + "sync" "testing" + "time" + + "github.com/launchdarkly/eventsource" "github.com/open-feature/flagd/core/pkg/logger" "github.com/open-feature/flagd/core/pkg/model" @@ -29,6 +33,45 @@ func mustSelector(t *testing.T, expr string) store.Selector { return s } +func newTestStore(t *testing.T) *store.Store { + t.Helper() + s, err := store.NewStore(logger.NewLogger(nil, false), []string{"src1", "src2"}) + require.NoError(t, err) + return s +} + +// newTracker builds a Tracker over a real eventsource server, so publishes are exercised. +func newTracker(t *testing.T, s store.IStore) *Tracker { + t.Helper() + es := eventsource.NewServer() + t.Cleanup(es.Close) + + tr := NewTracker(logger.NewLogger(nil, false), s, es) + t.Cleanup(tr.Close) + return tr +} + +// subscribe registers a channel and ties its release to the test cleanup. +func subscribe(t *testing.T, tr *Tracker, expr string) { + t.Helper() + release, err := tr.Subscribe(expr, mustSelector(t, expr)) + require.NoError(t, err) + t.Cleanup(release) +} + +// versionOf polls until the channel has a seeded ETag, since seeding is asynchronous. +func versionOf(t *testing.T, tr *Tracker, channel string) (string, int64) { + t.Helper() + var etag string + var lastModified int64 + require.Eventually(t, func() bool { + var ok bool + etag, lastModified, ok = tr.Version(channel) + return ok + }, 3*time.Second, 5*time.Millisecond, "no version seeded for channel %q", channel) + return etag, lastModified +} + func TestFingerprint_StableAndSensitive(t *testing.T) { a := testFlag("fs1", "a", "on") b := testFlag("fs1", "b", "off") @@ -48,113 +91,167 @@ func TestFingerprint_StableAndSensitive(t *testing.T) { assert.Equal(t, emptyGroup, nilGroup) } -func TestTracker_Update_ChangedChannels(t *testing.T) { - tr := &Tracker{versions: map[string]version{}} +// TestTracker_PassesSelectorToStore pins that filtering is delegated to the store by handing it +// the channel's selector, rather than re-implemented by grouping flags in memory. +func TestTracker_PassesSelectorToStore(t *testing.T) { + rec := &recordingStore{} + tr := newTracker(t, rec) - fs1a := testFlag("fs1", "a", "on") - fs2b := testFlag("fs2", "b", "on") + subscribe(t, tr, "flagSetId=fs1") - // first update seeds every channel (this is the init snapshot Run skips) - first := tr.update([]model.Flag{fs1a, fs2b}) - assert.Contains(t, first, allChannel) - assert.Contains(t, first, "fs1") - assert.Contains(t, first, "fs2") + selectors := rec.selectors() + require.Len(t, selectors, 1) + expected := mustSelector(t, "flagSetId=fs1") + assert.Equal(t, expected.ToLogString(), selectors[0].ToLogString()) +} - // no change -> no channels - assert.Empty(t, tr.update([]model.Flag{fs1a, fs2b})) +func TestTracker_SharesOneWatchPerChannel(t *testing.T) { + rec := &recordingStore{} + tr := newTracker(t, rec) - // change only fs1 -> catch-all + fs1, not fs2 - fs1aModified := testFlag("fs1", "a", "off") - changed := tr.update([]model.Flag{fs1aModified, fs2b}) - assert.Contains(t, changed, allChannel) - assert.Contains(t, changed, "fs1") - assert.NotContains(t, changed, "fs2") + subscribe(t, tr, "flagSetId=fs1") + subscribe(t, tr, "flagSetId=fs1") + assert.Len(t, rec.selectors(), 1, "subscribers on one channel must share a store watch") - // deleting a whole flag set notifies that set's channel + catch-all - removed := tr.update([]model.Flag{fs1aModified}) - assert.Contains(t, removed, allChannel) - assert.Contains(t, removed, "fs2") - assert.NotContains(t, removed, "fs1") + subscribe(t, tr, "flagSetId=fs2") + assert.Len(t, rec.selectors(), 2, "a distinct channel needs its own watch") } -func TestTracker_Update_NilFlagSetIdNotSubscribable(t *testing.T) { - tr := &Tracker{versions: map[string]version{}} +func TestTracker_Version_NoSubscription(t *testing.T) { + tr := newTracker(t, newTestStore(t)) - changed := tr.update([]model.Flag{testFlag(store.NilFlagSetId(), "a", "on")}) - assert.Contains(t, changed, allChannel, "catch-all must fire for flags without a flagSetId") - assert.NotContains(t, changed, store.NilFlagSetId(), "internal nilFlagSetId must not be a subscribable channel") + _, _, ok := tr.Version("flagSetId=fs1") + assert.False(t, ok, "no live stream for the channel means no ETag and no 304") } -func TestTracker_Run_NilStoreDoesNotPanic(t *testing.T) { - tr := NewTracker(logger.NewLogger(nil, false), nil, nil) - // Run must return promptly instead of dereferencing the nil store (which would panic in - // the background goroutine and terminate the process). - require.NotPanics(t, func() { tr.Run(context.Background()) }) +func TestTracker_ReleaseTearsDownLastSubscriberOnly(t *testing.T) { + tr := newTracker(t, newTestStore(t)) + + releaseA, err := tr.Subscribe("flagSetId=fs1", mustSelector(t, "flagSetId=fs1")) + require.NoError(t, err) + releaseB, err := tr.Subscribe("flagSetId=fs1", mustSelector(t, "flagSetId=fs1")) + require.NoError(t, err) + + releaseA() + assert.Len(t, tr.Channels(), 1, "one subscriber leaving must not tear down a shared channel") + + releaseB() + assert.Empty(t, tr.Channels()) + _, _, ok := tr.Version("flagSetId=fs1") + assert.False(t, ok) } -// closingStore closes the watcher immediately without emitting, simulating store.Watch's -// error-close path. +func TestTracker_IdenticalUpdateIsSuppressed(t *testing.T) { + s := newTestStore(t) + flags := []model.Flag{testFlag("fs1", "a", "on")} + s.Update("src1", flags, model.Metadata{"flagSetId": "fs1"}, false) + + tr := newTracker(t, s) + subscribe(t, tr, "flagSetId=fs1") + + _, firstModified := versionOf(t, tr, "flagSetId=fs1") + + // Update re-inserts unconditionally, so an identical re-sync still wakes the watch + s.Update("src1", flags, model.Metadata{"flagSetId": "fs1"}, false) + time.Sleep(200 * time.Millisecond) + + _, stillModified := versionOf(t, tr, "flagSetId=fs1") + assert.Equal(t, firstModified, stillModified, "lastModified must not move when nothing changed") +} + +func TestTracker_VersionTracksChanges(t *testing.T) { + s := newTestStore(t) + s.Update("src1", []model.Flag{testFlag("fs1", "a", "on")}, model.Metadata{"flagSetId": "fs1"}, false) + + tr := newTracker(t, s) + subscribe(t, tr, "flagSetId=fs1") + + before, _ := versionOf(t, tr, "flagSetId=fs1") + + s.Update("src1", []model.Flag{testFlag("fs1", "a", "off")}, model.Metadata{"flagSetId": "fs1"}, false) + require.Eventually(t, func() bool { + etag, _, _ := tr.Version("flagSetId=fs1") + return etag != before + }, 3*time.Second, 5*time.Millisecond, "the ETag must move when the config changes") +} + +// closingStore closes the watcher without emitting, simulating store.Watch's error-close path. type closingStore struct{} func (closingStore) Get(context.Context, string, *store.Selector) (model.Flag, model.Metadata, error) { return model.Flag{}, nil, nil } + func (closingStore) GetAll(context.Context, *store.Selector) ([]model.Flag, model.Metadata, error) { return nil, nil, nil } + func (closingStore) Watch(_ context.Context, _ *store.Selector, watcher chan<- store.FlagQueryResult) { close(watcher) } + func (closingStore) Update(string, []model.Flag, model.Metadata, bool) {} -func TestTracker_Run_UnexpectedCloseInvalidatesVersions(t *testing.T) { - tr := NewTracker(logger.NewLogger(nil, false), closingStore{}, nil) - tr.versions = map[string]version{allKey: {etag: "frozen"}} +func TestTracker_StoreErrorInvalidatesVersion(t *testing.T) { + tr := newTracker(t, closingStore{}) - // context is NOT cancelled -> the watcher closing is unexpected (store error path) - tr.Run(context.Background()) + subscribe(t, tr, "flagSetId=fs1") - _, _, ok := tr.Version(store.Selector{}) - assert.False(t, ok, "frozen versions must be invalidated so the bulk handler stops serving stale 304s") + assert.Eventually(t, func() bool { + _, _, ok := tr.Version("flagSetId=fs1") + return !ok + }, 3*time.Second, 5*time.Millisecond, + "a dead watch must stop serving its ETag, or 304s would be stale forever") } -func TestTracker_Run_ContextCancelKeepsVersions(t *testing.T) { - tr := NewTracker(logger.NewLogger(nil, false), closingStore{}, nil) - tr.versions = map[string]version{allKey: {etag: "current"}} +func TestTracker_CloseRejectsNewSubscribers(t *testing.T) { + es := eventsource.NewServer() + defer es.Close() + tr := NewTracker(logger.NewLogger(nil, false), newTestStore(t), es) + + release, err := tr.Subscribe("flagSetId=fs1", mustSelector(t, "flagSetId=fs1")) + require.NoError(t, err) - ctx, cancel := context.WithCancel(context.Background()) - cancel() // expected shutdown + require.NotPanics(t, tr.Close) + assert.Empty(t, tr.Channels()) - tr.Run(ctx) + _, err = tr.Subscribe("flagSetId=fs2", mustSelector(t, "flagSetId=fs2")) + assert.Error(t, err) - etag, _, ok := tr.Version(store.Selector{}) - assert.True(t, ok, "versions must be retained on a normal context-cancel shutdown") - assert.Equal(t, "current", etag) + assert.NotPanics(t, release) // releasing after Close is a no-op } -func TestTracker_Version(t *testing.T) { - tr := &Tracker{versions: map[string]version{}} - tr.update([]model.Flag{testFlag("fs1", "a", "on")}) +// recordingStore records the selector handed to each Watch call, mimicking the real store's +// contract: an immediate initial snapshot on a buffered channel, closed on ctx cancellation. +type recordingStore struct { + mu sync.Mutex + sels []store.Selector +} - // catch-all (empty selector) - allEtag, _, ok := tr.Version(store.Selector{}) - require.True(t, ok) - assert.NotEmpty(t, allEtag) +func (r *recordingStore) selectors() []store.Selector { + r.mu.Lock() + defer r.mu.Unlock() + return append([]store.Selector(nil), r.sels...) +} - // flagSetId selector - fsEtag, _, ok := tr.Version(mustSelector(t, "flagSetId=fs1")) - require.True(t, ok) - assert.NotEmpty(t, fsEtag) +func (r *recordingStore) Get(context.Context, string, *store.Selector) (model.Flag, model.Metadata, error) { + return model.Flag{}, nil, nil +} - // source selector is tracked for ETag lookups - _, _, ok = tr.Version(mustSelector(t, "source=src1")) - assert.True(t, ok) +func (r *recordingStore) GetAll(context.Context, *store.Selector) ([]model.Flag, model.Metadata, error) { + return nil, nil, nil +} - // unknown source -> not tracked - _, _, ok = tr.Version(mustSelector(t, "source=missing")) - assert.False(t, ok) +func (r *recordingStore) Watch(ctx context.Context, selector *store.Selector, watcher chan<- store.FlagQueryResult) { + r.mu.Lock() + r.sels = append(r.sels, *selector) + r.mu.Unlock() - // unknown flagSetId -> not tracked - _, _, ok = tr.Version(mustSelector(t, "flagSetId=nope")) - assert.False(t, ok) + go func() { + watcher <- store.FlagQueryResult{} + <-ctx.Done() + close(watcher) + }() } + +func (r *recordingStore) Update(string, []model.Flag, model.Metadata, bool) {} diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go b/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go index a4d1c48ad..8423f0ea1 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go @@ -13,7 +13,6 @@ import ( "github.com/open-feature/flagd/core/pkg/logger" "github.com/open-feature/flagd/core/pkg/model" "github.com/open-feature/flagd/core/pkg/service/ofrep" - "github.com/open-feature/flagd/core/pkg/store" svc "github.com/open-feature/flagd/flagd/pkg/service" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -26,7 +25,7 @@ type fakeVersioner struct { ok bool } -func (f fakeVersioner) Version(_ store.Selector) (string, int64, bool) { +func (f fakeVersioner) Version(_ string) (string, int64, bool) { return f.etag, f.lastModified, f.ok } @@ -157,7 +156,28 @@ func TestHandleBulkEvaluation_AdvertisesEventStreams(t *testing.T) { name: "flagSetId selector, origin omitted", selectorHeader: "flagSetId=fs1", wantOrigin: "", - wantRequestURI: "/ofrep/v1/sse?channels=fs1", + wantRequestURI: "/ofrep/v1/sse?channels=flagSetId%3Dfs1", + }, + { + // previously fell back to the catch-all channel, and so was woken by every change + name: "source selector gets its own channel", + selectorHeader: "source=src1", + wantOrigin: "", + wantRequestURI: "/ofrep/v1/sse?channels=source%3Dsrc1", + }, + { + // advertised verbatim, so the SSE endpoint parses it back to the same selector + name: "bare selector expression is advertised verbatim", + selectorHeader: "mySource", + wantOrigin: "", + wantRequestURI: "/ofrep/v1/sse?channels=mySource", + }, + { + // "flags with no flagSetId"; the internal nil flagSetId must never be advertised + name: "empty flagSetId selector", + selectorHeader: "flagSetId=", + wantOrigin: "", + wantRequestURI: "/ofrep/v1/sse?channels=flagSetId%3D", }, { name: "no selector (catch-all), origin omitted", @@ -170,7 +190,7 @@ func TestHandleBulkEvaluation_AdvertisesEventStreams(t *testing.T) { publicURL: "https://flags.example.com/", selectorHeader: "flagSetId=fs1", wantOrigin: "https://flags.example.com", - wantRequestURI: "/ofrep/v1/sse?channels=fs1", + wantRequestURI: "/ofrep/v1/sse?channels=flagSetId%3Dfs1", }, } @@ -226,3 +246,31 @@ func TestHandleBulkEvaluation_SSEDisabled_NoEventStreams(t *testing.T) { require.NoError(t, json.Unmarshal(recorder.Body.Bytes(), &resp)) assert.Empty(t, resp.EventStreams) } + +// TestHandleBulkEvaluation_NoLiveStream_NoETag pins the contract that follows from tracking +// versions per live subscription: with no stream open there is no cache validator to offer, so +// the response is a plain 200 that still advertises eventStreams. +func TestHandleBulkEvaluation_NoLiveStream_NoETag(t *testing.T) { + log := logger.NewLogger(nil, false) + eval := mock.NewMockIEvaluator(gomock.NewController(t)) + eval.EXPECT().ResolveAllValues(gomock.Any(), gomock.Any(), gomock.Any()). + Return([]evaluator.AnyValue{}, model.Metadata{}, nil) + + h := handler{ + Logger: log, + evaluator: eval, + versioner: fakeVersioner{ok: false}, + sseEnabled: true, + sseInactivityDelaySec: 120, + } + + recorder := serveBulk(h, newBulkRequest(t, "flagSetId=fs1", `"stale-etag"`)) + + require.Equal(t, http.StatusOK, recorder.Code, "without a tracked version we must not serve a 304") + assert.Empty(t, recorder.Header().Get("ETag")) + + var resp ofrep.BulkEvaluationResponse + require.NoError(t, json.Unmarshal(recorder.Body.Bytes(), &resp)) + require.Len(t, resp.EventStreams, 1) + assert.NotContains(t, resp.Metadata, "flagConfigLastModified") +} From a910b52f5890f55f1227accf8a54d44fca8cc459 Mon Sep 17 00:00:00 2001 From: Jamie Sinn Date: Thu, 20 Aug 2026 10:36:31 -0400 Subject: [PATCH 7/9] Apply suggestions from code review Co-authored-by: Roman Dmytrenko Signed-off-by: Jamie Sinn --- flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go index 2712de549..bb1e66903 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go @@ -134,8 +134,7 @@ func (t *Tracker) watch(ctx context.Context, channel string, sub *subscription, // serving its ETag, which would otherwise answer 304 from a frozen fingerprint forever. sub.ver.Store(nil) if t.logger != nil { - t.logger.Error(fmt.Sprintf( - "ofrep sse watch for channel %q ended unexpectedly; its cached ETag is invalidated", channel)) + t.logger.Error("ofrep sse watch ended unexpectedly; its cached ETag is invalidated", zap.String("channel", channel)) } } @@ -150,7 +149,7 @@ func (t *Tracker) publish(channel string, sub *subscription) { id := strconv.FormatInt(t.eventID.Add(1), 10) t.es.Publish([]string{channel}, newRefetchEvent(id, v.etag, v.lastModified)) if t.logger != nil { - t.logger.Debug(fmt.Sprintf("published refetch event to channel %q (etag=%s)", channel, v.etag)) + t.logger.Debug("published refetch event", zap.String("channel", channel), zap.String("etag", v.etag) } } From d52b2da36159a2b876d3eb01633201de6b1efc3e Mon Sep 17 00:00:00 2001 From: Jamie Sinn Date: Thu, 20 Aug 2026 11:20:03 -0400 Subject: [PATCH 8/9] resolve PR comments use selector for fingerprinting instead of flagset id fix some concurrency issues/potential race Signed-off-by: Jamie Sinn --- core/pkg/store/query.go | 5 -- .../flag-evaluation/ofrep/sse/tracker.go | 60 ++++++++----------- .../flag-evaluation/ofrep/sse/tracker_test.go | 21 ++++--- 3 files changed, 40 insertions(+), 46 deletions(-) diff --git a/core/pkg/store/query.go b/core/pkg/store/query.go index d62d19c22..fdd133bd4 100644 --- a/core/pkg/store/query.go +++ b/core/pkg/store/query.go @@ -28,11 +28,6 @@ const flagSetIdKeySourceCompoundIndex = flagSetIdIndex + "+" + keyIndex + "+" + // any flag without a "flagSetId" is assigned this one; it's never exposed externally var nilFlagSetId = uuid.New().String() -// NilFlagSetId returns the startup-generated flagSetId assigned to flags that have -// no explicit flagSetId. It is never exposed to clients, but consumers that group or -// fingerprint flags by flagSetId (e.g. the OFREP SSE tracker) need to recognise it. -func NilFlagSetId() string { return nilFlagSetId } - // A Selector represents a set of constraints used to query the store. type Selector struct { indexMap map[string]string diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go index bb1e66903..5b9cfa8da 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker.go @@ -16,6 +16,7 @@ import ( "github.com/open-feature/flagd/core/pkg/logger" "github.com/open-feature/flagd/core/pkg/model" "github.com/open-feature/flagd/core/pkg/store" + "go.uber.org/zap" ) type version struct { @@ -44,8 +45,7 @@ type Tracker struct { subs map[string]*subscription closed bool - wg sync.WaitGroup - eventID atomic.Int64 + wg sync.WaitGroup } func NewTracker(log *logger.Logger, s store.IStore, es *eventsource.Server) *Tracker { @@ -110,13 +110,14 @@ func (t *Tracker) watch(ctx context.Context, channel string, sub *subscription, // This loop must run until the channel closes: store.Watch's send does not select on // ctx.Done(), so abandoning the channel would leak the store's goroutine. for res := range watcher { - fp := fingerprint(res.Flags) - if prev := sub.ver.Load(); prev != nil && prev.etag == fp { + fp := fingerprint(&sub.selector, res.Flags) + prev := sub.ver.Load() + if prev != nil && prev.etag == fp { // A no-op wakeup: an identical re-sync, or a coarser radix watch channel firing // for a change outside this selector. continue } - sub.ver.Store(&version{etag: fp, lastModified: time.Now().Unix()}) + sub.ver.CompareAndSwap(prev, &version{etag: fp, lastModified: time.Now().Unix()}) if first { // seed only; the connecting client re-fetches unconditionally anyway (ADR-0008) @@ -146,10 +147,10 @@ func (t *Tracker) publish(channel string, sub *subscription) { return } - id := strconv.FormatInt(t.eventID.Add(1), 10) + id := strconv.FormatInt(time.Now().UnixMilli(), 10) t.es.Publish([]string{channel}, newRefetchEvent(id, v.etag, v.lastModified)) if t.logger != nil { - t.logger.Debug("published refetch event", zap.String("channel", channel), zap.String("etag", v.etag) + t.logger.Debug("published refetch event", zap.String("channel", channel), zap.String("etag", v.etag)) } } @@ -201,37 +202,28 @@ func (t *Tracker) Close() { t.wg.Wait() } -// fingerprint produces a deterministic, restart-stable hash of a group of flag definitions. -// nilFlagSetId is normalised to "" so identical config yields the same fingerprint across -// restarts. -func fingerprint(flags []model.Flag) string { - sorted := make([]model.Flag, len(flags)) - copy(sorted, flags) - sort.Slice(sorted, func(i, j int) bool { - if sorted[i].FlagSetId != sorted[j].FlagSetId { - return sorted[i].FlagSetId < sorted[j].FlagSetId - } - if sorted[i].Source != sorted[j].Source { - return sorted[i].Source < sorted[j].Source - } - return sorted[i].Key < sorted[j].Key - }) - - nilID := store.NilFlagSetId() - h := sha256.New() - for _, f := range sorted { - fsid := f.FlagSetId - if fsid == nilID { - fsid = "" - } - h.Write([]byte(fsid)) - h.Write([]byte{0}) +// fingerprint produces a deterministic hash of the result set a channel's watch resolves to. +func fingerprint(selector *store.Selector, flags []model.Flag) string { + digests := make([]string, 0, len(flags)) + for _, f := range flags { + h := sha256.New() h.Write([]byte(f.Key)) h.Write([]byte{0}) - // model.Flag.MarshalJSON is stable (map keys sorted) and covers the definition fields. + h.Write([]byte(f.Source)) + h.Write([]byte{0}) b, _ := json.Marshal(f) h.Write(b) - h.Write([]byte{0}) + digests = append(digests, string(h.Sum(nil))) + } + sort.Strings(digests) + + h := sha256.New() + // the selector path scopes the hash to this watch, so two channels resolving to the same + // flags never share an ETag unless they come from the same selector source. + h.Write([]byte(selector.ToLogString())) + h.Write([]byte{0}) + for _, d := range digests { + h.Write([]byte(d)) } return hex.EncodeToString(h.Sum(nil)) } diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go index d85e4f81b..cf99d3ae0 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/tracker_test.go @@ -73,22 +73,29 @@ func versionOf(t *testing.T, tr *Tracker, channel string) (string, int64) { } func TestFingerprint_StableAndSensitive(t *testing.T) { + sel := mustSelector(t, "flagSetId=fs1") a := testFlag("fs1", "a", "on") b := testFlag("fs1", "b", "off") // deterministic for identical input - assert.Equal(t, fingerprint([]model.Flag{a, b}), fingerprint([]model.Flag{a, b})) + assert.Equal(t, fingerprint(&sel, []model.Flag{a, b}), fingerprint(&sel, []model.Flag{a, b})) // order independent - assert.Equal(t, fingerprint([]model.Flag{a, b}), fingerprint([]model.Flag{b, a})) + assert.Equal(t, fingerprint(&sel, []model.Flag{a, b}), fingerprint(&sel, []model.Flag{b, a})) // sensitive to a definition change aModified := testFlag("fs1", "a", "off") - assert.NotEqual(t, fingerprint([]model.Flag{a}), fingerprint([]model.Flag{aModified})) + assert.NotEqual(t, fingerprint(&sel, []model.Flag{a}), fingerprint(&sel, []model.Flag{aModified})) - // nilFlagSetId is normalised so it does not leak the random UUID into the hash - nilGroup := fingerprint([]model.Flag{testFlag(store.NilFlagSetId(), "a", "on")}) - emptyGroup := fingerprint([]model.Flag{testFlag("", "a", "on")}) - assert.Equal(t, emptyGroup, nilGroup) + // duplicates are not collapsed: a flag appearing twice is a different result set + assert.NotEqual(t, fingerprint(&sel, []model.Flag{a}), fingerprint(&sel, []model.Flag{a, a})) + + // the flagSetId carried on a flag is not part of the identity; the selector the watch was + // opened with is + relabelled := testFlag("some-other-flag-set", "a", "on") + assert.Equal(t, fingerprint(&sel, []model.Flag{a}), fingerprint(&sel, []model.Flag{relabelled})) + + other := mustSelector(t, "source=src1") + assert.NotEqual(t, fingerprint(&sel, []model.Flag{a}), fingerprint(&other, []model.Flag{a})) } // TestTracker_PassesSelectorToStore pins that filtering is delegated to the store by handing it From 8c73c7ba1090f13bf25318cbaa17a53d7be89a80 Mon Sep 17 00:00:00 2001 From: Jamie Sinn Date: Thu, 20 Aug 2026 12:58:01 -0400 Subject: [PATCH 9/9] use path instead of channel param in the query string Signed-off-by: Jamie Sinn --- docs/reference/flagd-cli/flagd_start.md | 2 +- flagd/cmd/start.go | 2 +- .../service/flag-evaluation/ofrep/handler.go | 11 ++-- .../flag-evaluation/ofrep/ofrep_service.go | 2 +- .../flag-evaluation/ofrep/sse/handler.go | 16 +++-- .../flag-evaluation/ofrep/sse/service.go | 20 +++++++ .../flag-evaluation/ofrep/sse/service_test.go | 60 +++++++++++++++---- .../flag-evaluation/ofrep/sse_bulk_test.go | 17 ++++-- 8 files changed, 98 insertions(+), 32 deletions(-) diff --git a/docs/reference/flagd-cli/flagd_start.md b/docs/reference/flagd-cli/flagd_start.md index a36538875..4d98e9b4a 100644 --- a/docs/reference/flagd-cli/flagd_start.md +++ b/docs/reference/flagd-cli/flagd_start.md @@ -24,7 +24,7 @@ flagd start [flags] -R, --max-request-header int Maximum allowed request header size in bytes. Requests exceeding this are rejected with HTTP 431. Set to 0 to use Go's built-in default (1 MiB). WARNING: setting a very large or zero value may allow memory exhaustion from oversized headers. (default 1000000) -t, --metrics-exporter string Set the metrics exporter. Default(if unset) is Prometheus. Can be override to otel - OpenTelemetry metric exporter. Overriding to otel require otelCollectorURI to be present -r, --ofrep-port int32 ofrep service port (default 8016) - --ofrep-sse-enabled Enable the OFREP SSE change-notification endpoint (ADR-0008) at /ofrep/v1/sse on the ofrep port. Defaults to true. (default true) + --ofrep-sse-enabled Enable the OFREP SSE change-notification endpoint (ADR-0008) at /ofrep/v1/sse/{channel} on the ofrep port, where the channel is a selector expression. Defaults to true. (default true) --ofrep-sse-inactivity-delay int Inactivity delay (seconds) advertised to OFREP SSE clients in the eventStreams block. Clients close idle connections after this. Defaults to 120. (default 120) --ofrep-sse-public-url string Origin (scheme://host) advertised as the OFREP SSE eventStreams endpoint.origin. Omitted when empty, so clients resolve the requestUri against the OFREP base URL. Set when flagd is behind a proxy. -A, --otel-ca-path string tls certificate authority path to use with OpenTelemetry collector diff --git a/flagd/cmd/start.go b/flagd/cmd/start.go index b6de8aefc..d60faa832 100644 --- a/flagd/cmd/start.go +++ b/flagd/cmd/start.go @@ -61,7 +61,7 @@ func init() { flags.Int32P(syncPortFlagName, "g", 8015, "gRPC Sync port") flags.Int32P(ofrepPortFlagName, "r", 8016, "ofrep service port") - flags.Bool(ofrepSSEEnabledFlagName, true, "Enable the OFREP SSE change-notification endpoint (ADR-0008) at /ofrep/v1/sse on the ofrep port. Defaults to true.") + flags.Bool(ofrepSSEEnabledFlagName, true, "Enable the OFREP SSE change-notification endpoint (ADR-0008) at /ofrep/v1/sse/{channel} on the ofrep port, where the channel is a selector expression. Defaults to true.") flags.Int(ofrepSSEInactivityFlagName, 120, "Inactivity delay (seconds) advertised to OFREP SSE clients in the eventStreams block. Clients close idle connections after this. Defaults to 120.") flags.String(ofrepSSEPublicURLFlagName, "", "Origin (scheme://host) advertised as the OFREP SSE eventStreams endpoint.origin. Omitted when empty, so clients resolve the requestUri against the OFREP base URL. Set when flagd is behind a proxy.") flags.StringP(socketPathFlagName, "d", "", "Flagd unix socket path. "+ diff --git a/flagd/pkg/service/flag-evaluation/ofrep/handler.go b/flagd/pkg/service/flag-evaluation/ofrep/handler.go index ae3614b70..f90708828 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/handler.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/handler.go @@ -6,7 +6,6 @@ import ( "errors" "fmt" "net/http" - "net/url" "strings" "github.com/gorilla/mux" @@ -239,16 +238,14 @@ func (h *handler) bulkResponse(selectorExpression string, evaluations []evaluato } // eventStreams builds the ADR-0008 eventStreams advertisement pointing OFREP clients back at -// this flagd's SSE endpoint. The advertised channel is the request's own selector expression, so -// the stream covers exactly the flags the client just evaluated. +// this flagd's SSE endpoint. The advertised channel is the request's own selector expression, +// carried as the final path segment, so the stream covers exactly the flags the client just +// evaluated. // // It uses the structured `endpoint` form and omits origin unless a public URL is configured, so // the client resolves the requestUri against the OFREP base URL it is already talking to. func (h *handler) eventStreams(selectorExpression string) []ofrep.EventStream { - requestURI := ssePath - if selectorExpression != "" { - requestURI += "?" + sse.ChannelParam + "=" + url.QueryEscape(selectorExpression) - } + requestURI := sse.ChannelPath(ssePath, selectorExpression) return []ofrep.EventStream{{ Type: "sse", diff --git a/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service.go b/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service.go index ededd7fec..2e3a2febb 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/ofrep_service.go @@ -88,7 +88,7 @@ func NewOfrepService( evaluateHandler = http.MaxBytesHandler(evaluateHandler, cfg.MaxRequestBodyBytes) } if sseService != nil { - mux.Handle(ssePath, sseService.Handler()) + sseService.Register(mux, ssePath) } mux.Handle("/", evaluateHandler) diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/handler.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/handler.go index b70c113eb..0471c6808 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse/handler.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/handler.go @@ -6,20 +6,24 @@ import ( "github.com/open-feature/flagd/core/pkg/store" ) -// ChannelParam carries the selector expression the client wants change notifications for, using -// the same syntax as the Flagd-Selector header; empty selects every flag. It is exported so the -// bulk handler advertises the same parameter this handler reads. -const ChannelParam = "channels" +// channelPathVar is the path wildcard carrying the selector expression the client wants change +// notifications for, using the same syntax as the Flagd-Selector header. An empty channel (the +// bare stream path) selects every flag. See Service.Register for the routes it is bound to. +const channelPathVar = "channel" // Handler resolves the request's selector, takes a reference on the matching subscription so the // store watch stays alive, and streams events until the client disconnects. +// +// It must be mounted through Service.Register: the channel is read from the request path, so the +// route has to declare the channel wildcard. func (svc *Service) Handler() http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - channel := r.URL.Query().Get(ChannelParam) + // already percent-decoded by the mux, so a source containing "/" arrives intact + channel := r.PathValue(channelPathVar) selector, err := store.NewSelector(channel) if err != nil { // not echoing the expression back: it is unescaped client input - http.Error(w, "invalid selector in the 'channels' parameter", http.StatusBadRequest) + http.Error(w, "invalid selector in the channel path segment", http.StatusBadRequest) return } diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/service.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/service.go index 2e769e408..896fd890e 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse/service.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/service.go @@ -2,6 +2,8 @@ package sse import ( "context" + "net/http" + "net/url" "time" "github.com/launchdarkly/eventsource" @@ -51,6 +53,24 @@ func New(s store.IStore, cfg Config) *Service { // for conditional (ETag/304) evaluation. func (svc *Service) Tracker() *Tracker { return svc.tracker } +// Register mounts the stream on mux as {prefix}/{channel}, where the channel is a selector +// expression. The bare prefix is registered too, so subscribing to every flag does not depend on +// a trailing-slash redirect. +func (svc *Service) Register(mux *http.ServeMux, prefix string) { + h := svc.Handler() + mux.Handle(prefix, h) + mux.Handle(prefix+"/{"+channelPathVar+"}", h) +} + +// ChannelPath returns the stream path for a channel under prefix. The selector expression is a +// single path segment, so it is escaped: source selectors routinely contain "/". +func ChannelPath(prefix, channel string) string { + if channel == "" { + return prefix + } + return prefix + "/" + url.PathEscape(channel) +} + // Start runs the heartbeat loop until ctx is cancelled, then shuts down. It blocks, so it is // intended to run in its own goroutine. func (svc *Service) Start(ctx context.Context) error { diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse/service_test.go b/flagd/pkg/service/flag-evaluation/ofrep/sse/service_test.go index d8365ae20..b936891a1 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse/service_test.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse/service_test.go @@ -4,7 +4,6 @@ import ( "context" "net/http" "net/http/httptest" - "net/url" "testing" "time" @@ -16,6 +15,25 @@ import ( "github.com/stretchr/testify/require" ) +// testSSEPath mirrors the prefix the OFREP service mounts the stream on. +const testSSEPath = "/ofrep/v1/sse" + +// newStreamServer mounts the service the way the OFREP service does, so requests exercise the +// real routing: the channel is the final path segment. +func newStreamServer(t *testing.T, svc *Service) *httptest.Server { + t.Helper() + mux := http.NewServeMux() + svc.Register(mux, testSSEPath) + ts := httptest.NewServer(mux) + t.Cleanup(ts.Close) + return ts +} + +// streamURL builds the stream URL for a channel on ts. +func streamURL(ts *httptest.Server, channel string) string { + return ts.URL + ChannelPath(testSSEPath, channel) +} + // TestService_PublishesRefetchOnChange verifies the end-to-end wiring: a client subscribed with // a flagSetId selector receives an ADR-0008 refetchEvaluation event when that flag set changes. func TestService_PublishesRefetchOnChange(t *testing.T) { @@ -30,11 +48,10 @@ func TestService_PublishesRefetchOnChange(t *testing.T) { defer cancel() go func() { _ = svc.Start(ctx) }() - ts := httptest.NewServer(svc.Handler()) - defer ts.Close() + ts := newStreamServer(t, svc) // the channel token is a selector expression, same syntax as Flagd-Selector - stream, err := eventsource.Subscribe(ts.URL+"?channels="+url.QueryEscape("flagSetId=fs1"), "") + stream, err := eventsource.SubscribeWithURL(streamURL(ts, "flagSetId=fs1")) require.NoError(t, err) defer stream.Close() @@ -68,14 +85,13 @@ func TestService_SelectorScopedNotifications(t *testing.T) { defer cancel() go func() { _ = svc.Start(ctx) }() - ts := httptest.NewServer(svc.Handler()) - defer ts.Close() + ts := newStreamServer(t, svc) - bySource, err := eventsource.Subscribe(ts.URL+"?channels="+url.QueryEscape("source=src1"), "") + bySource, err := eventsource.SubscribeWithURL(streamURL(ts, "source=src1")) require.NoError(t, err) defer bySource.Close() - byFlagSet, err := eventsource.Subscribe(ts.URL+"?channels="+url.QueryEscape("flagSetId=fs2"), "") + byFlagSet, err := eventsource.SubscribeWithURL(streamURL(ts, "flagSetId=fs2")) require.NoError(t, err) defer byFlagSet.Close() @@ -103,13 +119,35 @@ func TestService_Handler_InvalidSelectorReturns400(t *testing.T) { require.NoError(t, err) svc := New(s, Config{Logger: log, HeartbeatInterval: time.Hour}) - ts := httptest.NewServer(svc.Handler()) - defer ts.Close() + ts := newStreamServer(t, svc) - resp, err := http.Get(ts.URL + "?channels=" + url.QueryEscape("bogusKey=1")) + resp, err := http.Get(streamURL(ts, "bogusKey=1")) require.NoError(t, err) defer resp.Body.Close() assert.Equal(t, http.StatusBadRequest, resp.StatusCode) assert.Empty(t, svc.Tracker().Channels(), "a rejected request must not create a subscription") } + +// TestService_ChannelComesFromPath pins the routing: the channel is the final path segment, the +// bare path is the catch-all, and a selector carrying a "/" survives the round trip escaped. +func TestService_ChannelComesFromPath(t *testing.T) { + log := logger.NewLogger(nil, false) + s, err := store.NewStore(log, []string{"./mySource"}) + require.NoError(t, err) + + svc := New(s, Config{Logger: log, HeartbeatInterval: time.Hour}) + ts := newStreamServer(t, svc) + + for _, channel := range []string{"", "flagSetId=fs1", "source=./mySource"} { + stream, err := eventsource.SubscribeWithURL(streamURL(ts, channel)) + require.NoError(t, err, "channel %q", channel) + defer stream.Close() + } + + assert.Eventually(t, func() bool { + return len(svc.Tracker().Channels()) == 3 + }, 3*time.Second, 5*time.Millisecond) + assert.ElementsMatch(t, []string{"", "flagSetId=fs1", "source=./mySource"}, svc.Tracker().Channels(), + "the subscription must key off the decoded path segment, not its escaped form") +} diff --git a/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go b/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go index 8423f0ea1..c0bb5e091 100644 --- a/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go +++ b/flagd/pkg/service/flag-evaluation/ofrep/sse_bulk_test.go @@ -156,28 +156,35 @@ func TestHandleBulkEvaluation_AdvertisesEventStreams(t *testing.T) { name: "flagSetId selector, origin omitted", selectorHeader: "flagSetId=fs1", wantOrigin: "", - wantRequestURI: "/ofrep/v1/sse?channels=flagSetId%3Dfs1", + wantRequestURI: "/ofrep/v1/sse/flagSetId=fs1", }, { // previously fell back to the catch-all channel, and so was woken by every change name: "source selector gets its own channel", selectorHeader: "source=src1", wantOrigin: "", - wantRequestURI: "/ofrep/v1/sse?channels=source%3Dsrc1", + wantRequestURI: "/ofrep/v1/sse/source=src1", }, { // advertised verbatim, so the SSE endpoint parses it back to the same selector name: "bare selector expression is advertised verbatim", selectorHeader: "mySource", wantOrigin: "", - wantRequestURI: "/ofrep/v1/sse?channels=mySource", + wantRequestURI: "/ofrep/v1/sse/mySource", + }, + { + // the channel is one path segment, so a source path must be escaped into it + name: "source containing a slash is path-escaped", + selectorHeader: "source=./mySource", + wantOrigin: "", + wantRequestURI: "/ofrep/v1/sse/source=.%2FmySource", }, { // "flags with no flagSetId"; the internal nil flagSetId must never be advertised name: "empty flagSetId selector", selectorHeader: "flagSetId=", wantOrigin: "", - wantRequestURI: "/ofrep/v1/sse?channels=flagSetId%3D", + wantRequestURI: "/ofrep/v1/sse/flagSetId=", }, { name: "no selector (catch-all), origin omitted", @@ -190,7 +197,7 @@ func TestHandleBulkEvaluation_AdvertisesEventStreams(t *testing.T) { publicURL: "https://flags.example.com/", selectorHeader: "flagSetId=fs1", wantOrigin: "https://flags.example.com", - wantRequestURI: "/ofrep/v1/sse?channels=flagSetId%3Dfs1", + wantRequestURI: "/ofrep/v1/sse/flagSetId=fs1", }, }