From 0eeaf8a77fe28d2db0e988cb6bd5fb9b015d7de5 Mon Sep 17 00:00:00 2001 From: Paul Wells Date: Sat, 12 Sep 2026 20:44:54 -0700 Subject: [PATCH 1/2] configutil: add Derive and a Validator hook on the observer Derive narrows an Observable onto a subtree of itself, so a package can define the config it owns and have an app embed it without that package importing the app's config type. project selects and must return a pointer into its argument: Load results are compared by identity. Validator lets a Builder reject a config before it is stored, so a reload is all-or-nothing and a bad edit leaves the running config in force. InitDefaults' error is now checked on the same path, where it was discarded. --- .changeset/configutil-derive.md | 6 ++ .changeset/configutil-validator.md | 6 ++ utils/configutil/derive.go | 45 ++++++++++ utils/configutil/derive_test.go | 127 +++++++++++++++++++++++++++++ utils/configutil/observer.go | 17 +++- utils/configutil/observer_test.go | 123 +++++++++++++++++++++++++++- 6 files changed, 321 insertions(+), 3 deletions(-) create mode 100644 .changeset/configutil-derive.md create mode 100644 .changeset/configutil-validator.md create mode 100644 utils/configutil/derive.go create mode 100644 utils/configutil/derive_test.go diff --git a/.changeset/configutil-derive.md b/.changeset/configutil-derive.md new file mode 100644 index 000000000..ba3fc2d9a --- /dev/null +++ b/.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 diff --git a/.changeset/configutil-validator.md b/.changeset/configutil-validator.md new file mode 100644 index 000000000..d7f43aa04 --- /dev/null +++ b/.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. diff --git a/utils/configutil/derive.go b/utils/configutil/derive.go new file mode 100644 index 000000000..18e1c396a --- /dev/null +++ b/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()) +} diff --git a/utils/configutil/derive_test.go b/utils/configutil/derive_test.go new file mode 100644 index 000000000..6b5d4dade --- /dev/null +++ b/utils/configutil/derive_test.go @@ -0,0 +1,127 @@ +// 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" + "go.uber.org/atomic" + + "github.com/livekit/protocol/utils/events" +) + +// *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 +} + +type stubObservable[T any] struct { + conf atomic.Pointer[T] + observers *events.ObserverList[*T] +} + +func newStubObservable[T any](conf *T) *stubObservable[T] { + s := &stubObservable[T]{observers: events.NewObserverList[*T](events.WithBlocking())} + s.conf.Store(conf) + return s +} + +func (s *stubObservable[T]) Observe(cb func(*T)) func() { return s.observers.On(cb) } + +func (s *stubObservable[T]) Load() *T { return s.conf.Load() } + +func (s *stubObservable[T]) emit(conf *T) { + s.conf.Store(conf) + s.observers.Emit(conf) +} + +func TestDerive(t *testing.T) { + src := newStubObservable(&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.emit(&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.emit(&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(newStubObservable(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 := newStubObservable(&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.emit(&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 := newStubObservable(&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.emit(&deriveAppConfig{Sweeper: deriveSweeperConfig{Period: 2 * time.Second}}) + require.Equal(t, 2*time.Second, period.Load()) +} diff --git a/utils/configutil/observer.go b/utils/configutil/observer.go index 6eb027a24..872225268 100644 --- a/utils/configutil/observer.go +++ b/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 @@ -164,7 +171,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) diff --git a/utils/configutil/observer_test.go b/utils/configutil/observer_test.go index ae02b59e2..95fc282c2 100644 --- a/utils/configutil/observer_test.go +++ b/utils/configutil/observer_test.go @@ -15,6 +15,7 @@ package configutil import ( + "errors" "os" "testing" "time" @@ -22,6 +23,7 @@ import ( "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 { From 7958464a6ea59f0b98e975d3cc04028193c26d53 Mon Sep 17 00:00:00 2001 From: Paul Wells Date: Sun, 13 Sep 2026 06:15:24 -0700 Subject: [PATCH 2/2] configutil: add NewStaticObserver Builds an Observer over an already-built config with no file to watch, for stubbing observable config in tests and for dev paths that must return the same *Observer the production path does. EmitConfigUpdate now stores the config it emits, so Load reflects an update pushed by hand rather than going stale. --- .changeset/configutil-static-observer.md | 6 ++++ utils/configutil/derive_test.go | 39 +++++------------------- utils/configutil/observer.go | 1 + utils/configutil/static.go | 29 ++++++++++++++++++ 4 files changed, 44 insertions(+), 31 deletions(-) create mode 100644 .changeset/configutil-static-observer.md create mode 100644 utils/configutil/static.go diff --git a/.changeset/configutil-static-observer.md b/.changeset/configutil-static-observer.md new file mode 100644 index 000000000..6064b75df --- /dev/null +++ b/.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. diff --git a/utils/configutil/derive_test.go b/utils/configutil/derive_test.go index 6b5d4dade..3ec61af0b 100644 --- a/utils/configutil/derive_test.go +++ b/utils/configutil/derive_test.go @@ -19,9 +19,6 @@ import ( "time" "github.com/stretchr/testify/require" - "go.uber.org/atomic" - - "github.com/livekit/protocol/utils/events" ) // *Observer is the observable Derive exists to narrow, so hold it to the @@ -45,28 +42,8 @@ type deriveInnerConfig struct { Name string } -type stubObservable[T any] struct { - conf atomic.Pointer[T] - observers *events.ObserverList[*T] -} - -func newStubObservable[T any](conf *T) *stubObservable[T] { - s := &stubObservable[T]{observers: events.NewObserverList[*T](events.WithBlocking())} - s.conf.Store(conf) - return s -} - -func (s *stubObservable[T]) Observe(cb func(*T)) func() { return s.observers.On(cb) } - -func (s *stubObservable[T]) Load() *T { return s.conf.Load() } - -func (s *stubObservable[T]) emit(conf *T) { - s.conf.Store(conf) - s.observers.Emit(conf) -} - func TestDerive(t *testing.T) { - src := newStubObservable(&deriveAppConfig{ + src := NewStaticObserver(&deriveAppConfig{ Sweeper: deriveSweeperConfig{Period: time.Second}, }) sweeper := Derive(src, func(c *deriveAppConfig) *deriveSweeperConfig { return &c.Sweeper }) @@ -78,26 +55,26 @@ func TestDerive(t *testing.T) { observed = append(observed, c.Period) }) - src.emit(&deriveAppConfig{Sweeper: deriveSweeperConfig{Period: 2 * time.Second}}) + 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.emit(&deriveAppConfig{Sweeper: deriveSweeperConfig{Period: 3 * time.Second}}) + 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(newStubObservable(conf), func(c *deriveAppConfig) *deriveSweeperConfig { return &c.Sweeper }) + 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 := newStubObservable(&deriveAppConfig{ + src := NewStaticObserver(&deriveAppConfig{ Nested: deriveNestedConfig{Inner: deriveInnerConfig{Name: "a"}}, }) nested := Derive(src, func(c *deriveAppConfig) *deriveNestedConfig { return &c.Nested }) @@ -108,13 +85,13 @@ func TestDeriveChained(t *testing.T) { done := make(chan string, 1) inner.Observe(func(c *deriveInnerConfig) { done <- c.Name }) - src.emit(&deriveAppConfig{Nested: deriveNestedConfig{Inner: deriveInnerConfig{Name: "b"}}}) + 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 := newStubObservable(&deriveAppConfig{ + src := NewStaticObserver(&deriveAppConfig{ Sweeper: deriveSweeperConfig{Period: time.Second}, }) sweeper := Derive(src, func(c *deriveAppConfig) *deriveSweeperConfig { return &c.Sweeper }) @@ -122,6 +99,6 @@ func TestDeriveFeedsAtomic(t *testing.T) { require.Equal(t, time.Second, period.Load()) - src.emit(&deriveAppConfig{Sweeper: deriveSweeperConfig{Period: 2 * time.Second}}) + src.EmitConfigUpdate(&deriveAppConfig{Sweeper: deriveSweeperConfig{Period: 2 * time.Second}}) require.Equal(t, 2*time.Second, period.Load()) } diff --git a/utils/configutil/observer.go b/utils/configutil/observer.go index 872225268..f16998819 100644 --- a/utils/configutil/observer.go +++ b/utils/configutil/observer.go @@ -84,6 +84,7 @@ func (c *Observer[T]) Close() { } func (c *Observer[T]) EmitConfigUpdate(conf *T) { + c.conf.Store(conf) c.observers.Emit(conf) } diff --git a/utils/configutil/static.go b/utils/configutil/static.go new file mode 100644 index 000000000..78e30e793 --- /dev/null +++ b/utils/configutil/static.go @@ -0,0 +1,29 @@ +// 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 ( + "github.com/livekit/protocol/utils/events" +) + +// NewStaticObserver stubs observable config in tests, and backs dev paths that +// must hand back the same *Observer the production path does. +// +// The nil builder is safe only because load is unreachable without a watcher. +func NewStaticObserver[T any](conf *T) *Observer[T] { + c := &Observer[T]{observers: events.NewObserverList[*T](events.WithBlocking())} + c.conf.Store(conf) + return c +}