SSE stream clients can refetch a stale snapshot because updates are published before they are stored

[markphelps]

Problem

A client that uses the SSE stream (/client/v2/environments/{env}/namespaces/{ns}/stream, or the OFREP _stream endpoint) can end up with a stale snapshot until the next flag change.

Cause

• Environment.updateSnapshot publishes the new snapshot to stream subscribers before it stores the snapshot that the snapshot endpoint serves (store.go:806, then store.go:815).
• Over SSE, the stream sends only a {"type":"refetchEvaluation","etag":...} hint, not the snapshot itself (gateway.go:124-132).
• A client that refetches as soon as it gets the hint reads e.snap.Load(), which can still be the old snapshot, or "namespace not found" if the namespace is new.
• The stream has already recorded the new digest as sent, so no second event follows.

The window is not only a race between two instructions:

• Publish waits on every subscriber before returning, with a 15s timeout per send (publisher.go).
• One slow client (full buffer or TCP backpressure) therefore delays Store by up to 15s, while faster subscribers have already been told to refetch.
• OFREP clients can't detect the problem by comparing etags, because the OFREP bulk etag is a hash of the results, not of the snapshot digest.

Native gRPC stream clients are not affected, because they receive the full snapshot.

Reproduce

This test fails with published digest="…" served digest="" getErr=evaluation snapshot for namespace "team-a" not found:

internal/storage/environments/git/publish_order_test.go

package git

import (
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.flipt.io/flipt/internal/config"
	"go.flipt.io/flipt/internal/storage/environments/evaluation"
	"go.flipt.io/flipt/internal/storage/environments/fs"
	storagegit "go.flipt.io/flipt/internal/storage/git"
	rpcenvironments "go.flipt.io/flipt/rpc/v2/environments"
	rpcevaluation "go.flipt.io/flipt/rpc/v2/evaluation"
	"go.uber.org/zap/zaptest"
)

// A subscriber notified of a new snapshot that immediately reads the
// environment's current snapshot (what the GET snapshot endpoint serves)
// should see the snapshot it was just notified about.
func Test_Environment_SnapshotStoredBeforePublish(t *testing.T) {
	logger := zaptest.NewLogger(t)
	ctx := t.Context()
	repo, err := storagegit.NewRepository(ctx, logger)
	require.NoError(t, err)

	publisher := evaluation.NewSnapshotPublisher(ctx, logger, evaluation.WithTimeout(5*time.Second))
	env, err := NewEnvironmentFromRepo(ctx, logger, &config.EnvironmentConfig{Name: "production"}, repo, fs.NewStorage(logger), publisher, config.TemplatesConfig{})
	require.NoError(t, err)

	const ns = "team-a"
	// sub1 = the SSE client; sub2 = another (slower) subscriber.
	ch1 := make(chan *rpcevaluation.EvaluationNamespaceSnapshot)
	ch2 := make(chan *rpcevaluation.EvaluationNamespaceSnapshot)
	c1, err := env.EvaluationNamespaceSnapshotSubscribe(ctx, ns, ch1)
	require.NoError(t, err)
	defer c1.Close()
	c2, err := env.EvaluationNamespaceSnapshotSubscribe(ctx, ns, ch2)
	require.NoError(t, err)
	defer c2.Close()

	rev := ""
	if resp, err := env.ListNamespaces(ctx); err == nil && resp != nil {
		rev = resp.Revision
	}
	_, err = env.CreateNamespace(ctx, rev, &rpcenvironments.Namespace{Key: ns, Name: "Team A"})
	require.NoError(t, err)

	type observed struct {
		published string
		served    string
		getErr    error
	}
	done := make(chan observed, 1)
	go func() {
		pub := <-ch1 // "refetchEvaluation" hint arrives
		// client immediately GETs the snapshot endpoint
		got, gerr := env.EvaluationNamespaceSnapshot(ctx, ns)
		o := observed{published: pub.Digest, getErr: gerr}
		if got != nil {
			o.served = got.Digest
		}
		<-ch2 // let the other subscriber drain so Publish can return
		done <- o
	}()

	require.NoError(t, env.updateSnapshot(ctx))

	o := <-done
	t.Logf("published digest=%q served digest=%q getErr=%v", o.published, o.served, o.getErr)
	require.NoError(t, o.getErr, "GET after publish should serve the published snapshot")
	assert.Equal(t, o.published, o.served, "GET after publish served a stale snapshot")
}

Suggested fix

In updateSnapshot, store the snapshot (and head) before calling publisher.Publish(snap). It may also be worth publishing asynchronously, so that a slow subscriber can't delay the swap.

[github-actions]

Issue Health Check: Well scoped

Type: Bug report

An exceptionally detailed bug report that clearly identifies a race condition in SSE snapshot publishing. It explains the root cause with specific code references, provides a failing reproduction test, and suggests a concrete fix. A maintainer can act on this immediately.

Thanks — this issue looks well scoped and ready for the maintainers to evaluate.

Need help with Flipt? Support the project through GitHub Sponsors or learn about Flipt Pro.