feature: Auto-cleanup derived refreshable subscriptions via runtime.AddCleanup (#435)

* feature: Auto-cleanup derived refreshable subscriptions via runtime.AddCleanup

Introduce cleanupState to coordinate GC and subscriber lifecycles for
derived refreshables. Upstream subscriptions are removed only when the
wrapper is GC'd AND all subscribers have unsubscribed, fixing a
regression where fire-and-forget subscriptions on derived refreshables
silently stopped receiving updates.

Add MapValues helper for map-typed refreshables with per-key lifecycle
management including context cancellation on key removal.

Add benchmarks demonstrating that parent.Update cost stays constant
regardless of how many derived refreshables are dropped (previously
leaked O(N) subscribers on master).

* Replace unwrapValidated type assertion with push-based value propagation

MergeValidated and CollectValidatedMutable previously used unwrapValidated
to extract the inner Validated from derivedValidated wrappers before capturing
them in doUpdate closures. This avoided reference cycles that would prevent
runtime.AddCleanup from firing, but required a type assertion on the private
derivedValidated type.

The new approach eliminates the type assertion entirely: subscriber callbacks
push extracted values (plain T and error) into shared mutable state protected
by a sync.Mutex, and doUpdate reads from that shared state instead of from
captured Validated[T] interface values. Since doUpdate never holds a reference
to any Validated (wrapper or inner), no reference cycle can form.

Also replaces atomic.Value with atomic.Pointer[T] in defaultRefreshable,
removing the remaining type assertions in the package.

* go fix

* Added a gcCollect() helper with a doc comment explaining why two GC cycles are needed (stale register/stack references can prevent collection on the first pass).

* comments

Author: bmoylan

refreshable/benchmark_test.go

@@ -0,0 +1,66 @@
+// Copyright (c) 2026 Palantir Technologies. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package refreshable_test
+
+import (
+	"fmt"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/palantir/pkg/refreshable/v2"
+)
+
+// BenchmarkParentUpdateAfterDerivedDrop measures the cost of parent.Update()
+// after creating and dropping N derived refreshables without calling stop().
+// Without runtime.AddCleanup, each dropped derived leaks a subscriber, so
+// Update iterates all N callbacks. With cleanup, GC removes them.
+func BenchmarkParentUpdateAfterDerivedDrop(b *testing.B) {
+	for _, dropped := range []int{0, 100, 1000, 10000} {
+		b.Run(fmt.Sprintf("dropped=%d", dropped), func(b *testing.B) {
+			parent := refreshable.New(0)
+			for i := range dropped {
+				derived := refreshable.MapAuto(parent, func(v int) int { return v + i })
+				_ = derived.Current()
+			}
+			gcCollect()
+			time.Sleep(100 * time.Millisecond)
+			b.ResetTimer()
+			for i := range b.N {
+				parent.Update(i + 1)
+			}
+		})
+	}
+}
+
+// BenchmarkParentSubscriberCountAfterDrop verifies that the parent's subscriber
+// list does not grow when derived refreshables are dropped and GC'd.
+// This is a correctness-flavored benchmark: without runtime.AddCleanup,
+// subscriberCount == dropped; with cleanup, subscriberCount == 0 after GC.
+func BenchmarkParentSubscriberCountAfterDrop(b *testing.B) {
+	for _, dropped := range []int{100, 1000} {
+		b.Run(fmt.Sprintf("dropped=%d", dropped), func(b *testing.B) {
+			for range b.N {
+				parent := refreshable.New(0)
+				var callCount atomic.Int64
+				parent.Subscribe(func(int) { callCount.Add(1) })
+				callCount.Store(0) // reset after initial Subscribe callback
+
+				for i := range dropped {
+					derived := refreshable.MapAuto(parent, func(v int) int { return v + i })
+					_ = derived.Current()
+				}
+				gcCollect()
+				time.Sleep(100 * time.Millisecond)
+
+				callCount.Store(0)
+				parent.Update(42)
+				if got := callCount.Load(); got != 1 {
+					b.Fatalf("expected 1 subscriber callback, got %d (leaked %d)", got, got-1)
+				}
+			}
+		})
+	}
+}

refreshable/derived.go

@@ -0,0 +1,132 @@
+// Copyright (c) 2026 Palantir Technologies. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package refreshable
+
+import (
+	"runtime"
+	"sync"
+	"sync/atomic"
+)
+
+// derivedRefreshable wraps a Refreshable and ties upstream subscription lifetime
+// to its own GC reachability and active subscriber count via cleanupState.
+//
+// Must be a separate allocation from the inner Refreshable: the inner is
+// typically captured by upstream subscription callbacks, so combining them
+// would create a reference cycle that prevents runtime.AddCleanup from firing.
+//
+// Upstream derived wrappers do NOT need to be kept alive by this wrapper.
+// When an upstream wrapper is collected, its cleanupState.gcDone is set but
+// cleanup is deferred until subCount reaches zero. The subscription callbacks
+// reference the upstream inner (not the wrapper), so updates continue to flow.
+// When the downstream eventually unsubscribes, the deferred cleanup fires and
+// correctly propagates the unsubscribe to the grandparent.
+type derivedRefreshable[T any] struct {
+	inner Refreshable[T]
+	state *cleanupState
+}
+
+func newDerivedRefreshable[T any](inner Refreshable[T], unsubs ...UnsubscribeFunc) *derivedRefreshable[T] {
+	state := &cleanupState{
+		unsubs: unsubs,
+	}
+	d := &derivedRefreshable[T]{
+		inner: inner,
+		state: state,
+	}
+	runtime.AddCleanup(d, (*cleanupState).markGCd, state)
+	return d
+}
+
+func (d *derivedRefreshable[T]) Current() T {
+	return d.inner.Current()
+}
+
+func (d *derivedRefreshable[T]) Subscribe(consumer func(T)) UnsubscribeFunc {
+	d.state.addSubscriber()
+	innerUnsub := d.inner.Subscribe(consumer)
+	state := d.state // capture state not d, so d remains GC-eligible
+	return sync.OnceFunc(func() {
+		innerUnsub()
+		state.removeSubscriber()
+	})
+}
+
+// derivedValidated wraps a Validated with the same GC cleanup semantics as
+// derivedRefreshable. See derivedRefreshable for design rationale.
+type derivedValidated[T any] struct {
+	inner Validated[T]
+	state *cleanupState
+}
+
+func newDerivedValidated[T any](inner Validated[T], unsubs ...UnsubscribeFunc) *derivedValidated[T] {
+	state := &cleanupState{
+		unsubs: unsubs,
+	}
+	d := &derivedValidated[T]{
+		inner: inner,
+		state: state,
+	}
+	runtime.AddCleanup(d, (*cleanupState).markGCd, state)
+	return d
+}
+
+func (d *derivedValidated[T]) Unvalidated() T {
+	return d.inner.Unvalidated()
+}
+
+func (d *derivedValidated[T]) Validation() (T, error) {
+	return d.inner.Validation()
+}
+
+func (d *derivedValidated[T]) SubscribeValidated(consumer func(Validated[T])) UnsubscribeFunc {
+	d.state.addSubscriber()
+	innerUnsub := d.inner.SubscribeValidated(consumer)
+	state := d.state // capture state not d, so d remains GC-eligible
+	return sync.OnceFunc(func() {
+		innerUnsub()
+		state.removeSubscriber()
+	})
+}
+
+// cleanupState coordinates between GC and active subscribers. Upstream
+// unsubscribe functions fire only when the derived wrapper is GC'd AND
+// the subscriber count reaches zero, allowing subscriptions on a derived
+// refreshable to outlive the derived reference itself.
+type cleanupState struct {
+	subCount atomic.Int32      // incremented by Subscribe, decremented by its returned unsub
+	gcDone   atomic.Bool       // set by runtime.AddCleanup callback
+	unsubs   []UnsubscribeFunc // upstream unsubscribe functions
+	once     sync.Once         // ensures cleanup runs exactly once
+}
+
+func (s *cleanupState) addSubscriber() {
+	s.subCount.Add(1)
+}
+
+func (s *cleanupState) removeSubscriber() {
+	s.subCount.Add(-1)
+	s.tryCleanup()
+}
+
+func (s *cleanupState) markGCd() {
+	s.gcDone.Store(true)
+	s.tryCleanup()
+}
+
+func (s *cleanupState) tryCleanup() {
+	if s.gcDone.Load() && s.subCount.Load() <= 0 {
+		s.once.Do(func() {
+			for _, unsub := range s.unsubs {
+				unsub()
+			}
+			// Release references captured by unsub closures so that objects
+			// reachable only through these closures (e.g. transport chains,
+			// upstream inners) become unreachable immediately rather than
+			// waiting for this cleanupState to be garbage collected.
+			s.unsubs = nil
+		})
+	}
+}