configutil: add Derive and a Validator hook on the observer

.changeset/configutil-derive.md

@@ -0,0 +1,6 @@
+---
+"github.com/livekit/protocol": patch
+"@livekit/protocol": patch
+---
+
+Add `configutil.Derive` for narrowing an observable config onto a subtree

.changeset/configutil-static-observer.md

@@ -0,0 +1,6 @@
+---
+"github.com/livekit/protocol": patch
+"@livekit/protocol": patch
+---
+
+Add `configutil.NewStaticObserver`, which builds an `Observer` over an already-built config with no file to watch, for stubbing observable config in tests. `EmitConfigUpdate` now also stores the config it emits, so `Load` reflects an update pushed by hand.

.changeset/configutil-validator.md

@@ -0,0 +1,6 @@
+---
+"github.com/livekit/protocol": patch
+"@livekit/protocol": patch
+---
+
+Add `configutil.Validator`, an optional builder hook run on every config load after `InitDefaults`. A config that fails validation, or whose `InitDefaults` returns an error, now fails `NewObserver`; on reload the failure is logged and the previous config stays in effect.

utils/configutil/derive.go

@@ -0,0 +1,45 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package configutil
+
+// Derive narrows an observable onto a subtree of itself, so a package can
+// define the config it owns, have the app embed it, and observe just that
+// subtree without importing the app's config package.
+//
+// project selects; it must not compute. It runs on every Load, so it has to be
+// cheap, and it must return a pointer into its argument rather than a freshly
+// built value — callers (NewAtomicPointer among them) compare what Load
+// returns by identity. For a projection that has to build something, use the
+// NewAtomic* helpers instead: they cache the result and recompute it on reload.
+//
+// The result holds no state and needs no cleanup: it subscribes to src only
+// while something is subscribed to it, and forwards src's emit semantics
+// unchanged.
+func Derive[Src, Dst any](src Observable[Src], project func(*Src) *Dst) Observable[Dst] {
+	return derived[Src, Dst]{src: src, project: project}
+}
+
+type derived[Src, Dst any] struct {
+	src     Observable[Src]
+	project func(*Src) *Dst
+}
+
+func (d derived[Src, Dst]) Observe(cb func(*Dst)) func() {
+	return d.src.Observe(func(c *Src) { cb(d.project(c)) })
+}
+
+func (d derived[Src, Dst]) Load() *Dst {
+	return d.project(d.src.Load())
+}

utils/configutil/derive_test.go

@@ -0,0 +1,104 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package configutil
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+// *Observer is the observable Derive exists to narrow, so hold it to the
+// interface here rather than discovering a mismatch at a call site.
+var _ Observable[deriveAppConfig] = (*Observer[deriveAppConfig])(nil)
+
+type deriveAppConfig struct {
+	Sweeper deriveSweeperConfig
+	Nested  deriveNestedConfig
+}
+
+type deriveSweeperConfig struct {
+	Period time.Duration
+}
+
+type deriveNestedConfig struct {
+	Inner deriveInnerConfig
+}
+
+type deriveInnerConfig struct {
+	Name string
+}
+
+func TestDerive(t *testing.T) {
+	src := NewStaticObserver(&deriveAppConfig{
+		Sweeper: deriveSweeperConfig{Period: time.Second},
+	})
+	sweeper := Derive(src, func(c *deriveAppConfig) *deriveSweeperConfig { return &c.Sweeper })
+
+	require.Equal(t, time.Second, sweeper.Load().Period)
+
+	var observed []time.Duration
+	unsubscribe := sweeper.Observe(func(c *deriveSweeperConfig) {
+		observed = append(observed, c.Period)
+	})
+
+	src.EmitConfigUpdate(&deriveAppConfig{Sweeper: deriveSweeperConfig{Period: 2 * time.Second}})
+	require.Equal(t, 2*time.Second, sweeper.Load().Period)
+	require.Equal(t, []time.Duration{2 * time.Second}, observed)
+
+	unsubscribe()
+	src.EmitConfigUpdate(&deriveAppConfig{Sweeper: deriveSweeperConfig{Period: 3 * time.Second}})
+	require.Equal(t, 3*time.Second, sweeper.Load().Period)
+	require.Equal(t, []time.Duration{2 * time.Second}, observed, "unsubscribed callback still fired")
+}
+
+func TestDeriveLoadIdentity(t *testing.T) {
+	conf := &deriveAppConfig{Sweeper: deriveSweeperConfig{Period: time.Second}}
+	sweeper := Derive(NewStaticObserver(conf), func(c *deriveAppConfig) *deriveSweeperConfig { return &c.Sweeper })
+
+	require.Same(t, &conf.Sweeper, sweeper.Load())
+	require.Same(t, sweeper.Load(), sweeper.Load())
+}
+
+func TestDeriveChained(t *testing.T) {
+	src := NewStaticObserver(&deriveAppConfig{
+		Nested: deriveNestedConfig{Inner: deriveInnerConfig{Name: "a"}},
+	})
+	nested := Derive(src, func(c *deriveAppConfig) *deriveNestedConfig { return &c.Nested })
+	inner := Derive(nested, func(c *deriveNestedConfig) *deriveInnerConfig { return &c.Inner })
+
+	require.Equal(t, "a", inner.Load().Name)
+
+	done := make(chan string, 1)
+	inner.Observe(func(c *deriveInnerConfig) { done <- c.Name })
+
+	src.EmitConfigUpdate(&deriveAppConfig{Nested: deriveNestedConfig{Inner: deriveInnerConfig{Name: "b"}}})
+	require.Equal(t, "b", <-done)
+	require.Equal(t, "b", inner.Load().Name)
+}
+
+func TestDeriveFeedsAtomic(t *testing.T) {
+	src := NewStaticObserver(&deriveAppConfig{
+		Sweeper: deriveSweeperConfig{Period: time.Second},
+	})
+	sweeper := Derive(src, func(c *deriveAppConfig) *deriveSweeperConfig { return &c.Sweeper })
+	period := NewAtomicDuration(sweeper, func(c *deriveSweeperConfig) time.Duration { return c.Period })
+
+	require.Equal(t, time.Second, period.Load())
+
+	src.EmitConfigUpdate(&deriveAppConfig{Sweeper: deriveSweeperConfig{Period: 2 * time.Second}})
+	require.Equal(t, 2*time.Second, period.Load())
+}

utils/configutil/observer.go

@@ -37,6 +37,13 @@ type Defaulter[T any] interface {
 	InitDefaults(*T) error
 }
 
+// Validator runs after InitDefaults on every load. A config that fails
+// validation is never stored or emitted, so on reload the previous config
+// stays in effect.
+type Validator[T any] interface {
+	Validate(*T) error
+}
+
 type Observer[T any] struct {
 	builder   Builder[T]
 	watcher   *fsnotify.Watcher
@@ -77,6 +84,7 @@ func (c *Observer[T]) Close() {
 }
 
 func (c *Observer[T]) EmitConfigUpdate(conf *T) {
+	c.conf.Store(conf)
 	c.observers.Emit(conf)
 }
 
@@ -164,7 +172,15 @@ func (c *Observer[T]) load(path string) (conf *T, err error) {
 	}
 
 	if d, ok := c.builder.(Defaulter[T]); ok {
-		d.InitDefaults(conf)
+		if err := d.InitDefaults(conf); err != nil {
+			return nil, fmt.Errorf("cannot apply config defaults: %w", err)
+		}
+	}
+
+	if v, ok := c.builder.(Validator[T]); ok {
+		if err := v.Validate(conf); err != nil {
+			return nil, fmt.Errorf("invalid config: %w", err)
+		}
 	}
 
 	c.conf.Store(conf)

utils/configutil/observer_test.go

@@ -15,13 +15,15 @@
 package configutil
 
 import (
+	"errors"
 	"os"
 	"testing"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
 )
 
 const testConfig0 = `foo: a`
@@ -34,17 +36,41 @@ type TestConfig struct {
 	Bar string `yaml:"bar"`
 }
 
-type testConfigBuilder struct{}
+var (
+	_ Defaulter[TestConfig] = testConfigBuilder{}
+	_ Validator[TestConfig] = testConfigBuilder{}
+)
+
+type testConfigBuilder struct {
+	initErr  error
+	validate func(*TestConfig) error
+}
 
 func (testConfigBuilder) New() (*TestConfig, error) {
 	return &TestConfig{}, nil
 }
 
-func (testConfigBuilder) InitDefaults(c *TestConfig) error {
+func (b testConfigBuilder) InitDefaults(c *TestConfig) error {
+	if b.initErr != nil {
+		return b.initErr
+	}
 	c.Bar = "c"
 	return nil
 }
 
+func (b testConfigBuilder) Validate(c *TestConfig) error {
+	if b.validate == nil {
+		return nil
+	}
+	return b.validate(c)
+}
+
+type newOnlyBuilder struct{}
+
+func (newOnlyBuilder) New() (*TestConfig, error) {
+	return &TestConfig{Foo: "new"}, nil
+}
+
 func TestConfigObserver(t *testing.T) {
 	f, err := os.CreateTemp(os.TempDir(), "lk-test-*.yaml")
 	t.Cleanup(func() {
@@ -128,6 +154,99 @@ func TestConfigObserver(t *testing.T) {
 	require.Zero(t, gaugeVecValue(promConfigLoadState, f.Name()))
 }
 
+const testConfigRejected = `foo: x`
+
+var errRejected = errors.New("rejected")
+
+func rejectFooX(c *TestConfig) error {
+	if c.Foo == "x" {
+		return errRejected
+	}
+	return nil
+}
+
+func TestConfigObserverValidation(t *testing.T) {
+	f, err := os.CreateTemp(os.TempDir(), "lk-test-*.yaml")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = f.Close() })
+	_, err = f.WriteString(testConfig0)
+	require.NoError(t, err)
+
+	obs, conf, err := NewObserver(f.Name(), testConfigBuilder{validate: rejectFooX})
+	require.NoError(t, err)
+	t.Cleanup(obs.Close)
+	require.Equal(t, "a", conf.Foo)
+
+	var emitted atomic.Int32
+	obs.Observe(func(*TestConfig) { emitted.Inc() })
+
+	_, err = f.WriteAt([]byte(testConfigRejected), 0)
+	require.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		return counterVecValue(promConfigReloadTotal, f.Name(), "failure") == 1
+	}, time.Second, 5*time.Millisecond)
+
+	// the rejected config is neither stored nor emitted, and the hash still
+	// reflects the config in use
+	require.Equal(t, "a", obs.Load().Foo)
+	require.Zero(t, emitted.Load())
+	require.Equal(t, float64(1), gaugeVecValue(promConfigLoadState, f.Name()))
+	require.Equal(t,
+		float64(configHash([]byte(testConfig0))),
+		gaugeVecValue(promConfigHash, f.Name()),
+	)
+
+	_, err = f.WriteAt([]byte(testConfig1), 0)
+	require.NoError(t, err)
+
+	require.Eventually(t, func() bool { return emitted.Load() == 1 }, time.Second, 5*time.Millisecond)
+	require.Equal(t, "b", obs.Load().Foo)
+	require.Zero(t, gaugeVecValue(promConfigLoadState, f.Name()))
+}
+
+func TestNewObserverLoadErrors(t *testing.T) {
+	f, err := os.CreateTemp(os.TempDir(), "lk-test-*.yaml")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = f.Close() })
+	_, err = f.WriteString(testConfig0)
+	require.NoError(t, err)
+
+	errDefaults := errors.New("defaults")
+	rejectAll := func(*TestConfig) error { return errRejected }
+
+	for _, tc := range []struct {
+		name    string
+		builder testConfigBuilder
+		want    error
+	}{
+		{"validate", testConfigBuilder{validate: rejectAll}, errRejected},
+		{"init_defaults", testConfigBuilder{initErr: errDefaults}, errDefaults},
+	} {
+		t.Run(tc.name+"/file", func(t *testing.T) {
+			obs, conf, err := NewObserver(f.Name(), tc.builder)
+			require.ErrorIs(t, err, tc.want)
+			require.Nil(t, obs)
+			require.Nil(t, conf)
+			require.Equal(t, float64(1), gaugeVecValue(promConfigLoadState, f.Name()))
+		})
+		t.Run(tc.name+"/nofile", func(t *testing.T) {
+			obs, conf, err := NewObserver("", tc.builder)
+			require.ErrorIs(t, err, tc.want)
+			require.Nil(t, obs)
+			require.Nil(t, conf)
+		})
+	}
+}
+
+func TestNewObserverBuilderOnly(t *testing.T) {
+	obs, conf, err := NewObserver("", newOnlyBuilder{})
+	require.NoError(t, err)
+	t.Cleanup(obs.Close)
+	require.Equal(t, &TestConfig{Foo: "new"}, conf)
+	require.Same(t, conf, obs.Load())
+}
+
 func gaugeVecValue(g *prometheus.GaugeVec, labels ...string) float64 {
 	m, err := g.GetMetricWithLabelValues(labels...)
 	if err != nil {