unbounded/backend: address Copilot review on #501

Four substantive findings + one stale comment (settings.go vpn.Init…
reference) answered via reply.

1. PatchSettings now dispatches settings.UnboundedKey to
   unbounded.SetEnabled the same way it dispatches PeerShareEnabledKey
   to applyPeerShare. The UI toggle now starts/stops the widget proxy
   immediately rather than persisting the value and waiting for the
   next NewConfigEvent.

2. Close now invokes unbounded.Stop(...) before r.cancel(). unbounded
   workers run on a context.Background-derived ctx (they must outlive
   any single NewConfigEvent), so without an explicit shutdown hook
   the broflake widget goroutine survived backend close. Uses a fresh
   5s-bounded ctx so a cancelled shutdown path doesn't skip the Stop.

3. SetEnabled now respects the full three-condition predicate before
   starting. Previously it called manager.start whenever cached
   *UnboundedConfig was non-nil, ignoring the cached
   Features[UNBOUNDED] flag — meaning a local toggle could start the
   proxy even when the server had said don't. Added lastFeatureOn to
   unboundedManager (set alongside lastCfg on every NewConfigEvent)
   and a shouldStart() method that re-checks all three conditions.
   SetEnabled now uses shouldStart(); the standalone
   shouldRunUnbounded function is removed (its job is now done by the
   method).

4. unbounded.ConnectionEvent shape aligned with peer.ConnectionEvent:
   {State, Source, Timestamp}. WorkerIdx is dropped from the wire
   shape (it remains in the per-callback Debug log for diagnostics
   but isn't part of the event contract). Doc comment updated to
   spell out the field semantics. Consumers that need to pair
   accept/close events for the same arc now key off Source (or
   arrival sequence within a single connection's lifetime, which is
   what the globe currently does).

The fifth Copilot comment cited a 'vpn.InitUnboundedSubscription'
reference in common/settings/settings.go:75 that no longer exists —
the consolidation cascade rewrote that doc to say 'the widget proxy
starts' without naming the lifecycle function. Answered in the thread
rather than re-edited.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: myleshorton

backend/radiance.go

@@ -326,6 +326,17 @@ func (r *LocalBackend) Close() {
 	r.closeOnce.Do(func() {
 		slog.Debug("Closing Radiance")
 		r.closePeerClient()
+		// unbounded.start spawns its worker on a context.Background-
+		// derived ctx (it has to outlive any single NewConfigEvent),
+		// so Close has to explicitly tell it to shut down — otherwise
+		// the broflake widget goroutine survives backend close and
+		// leaks until process exit. Use a fresh ctx so a cancelled
+		// shutdown path doesn't skip the Stop.
+		stopCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		if err := unbounded.Stop(stopCtx); err != nil {
+			slog.Warn("unbounded stop on backend close returned error", "error", err)
+		}
+		cancel()
 		// vpnClient is always set in production via NewLocalBackend, but
 		// peer-focused unit tests construct partial LocalBackends without
 		// one. Guard the call so Close stays robust under those paths
@@ -510,6 +521,16 @@ func (r *LocalBackend) PatchSettings(updates settings.Settings) error {
 		}
 	}
 
+	// Drive the Unbounded widget proxy off the toggle change immediately
+	// rather than waiting for the next NewConfigEvent to re-evaluate.
+	// SetEnabled is internally idempotent and checks the cached server
+	// feature flag + config before actually starting the worker.
+	if _, ok := diff[settings.UnboundedKey]; ok {
+		if err := unbounded.SetEnabled(settings.GetBool(settings.UnboundedKey)); err != nil {
+			slog.Warn("unbounded toggle failed", "error", err)
+		}
+	}
+
 	return nil
 }
 

unbounded/unbounded.go

@@ -28,6 +28,7 @@ import (
 	"log/slog"
 	"net"
 	"sync"
+	"time"
 
 	C "github.com/getlantern/common"
 
@@ -40,28 +41,44 @@ import (
 
 // ConnectionEvent fires every time a consumer (i.e. a censored client
 // being routed through this widget proxy) connects or disconnects via
-// the broflake mesh. State: +1 on accept, -1 on close. WorkerIdx is
-// broflake's internal worker slot identifier — used by the Flutter
-// globe to pair connect/disconnect events for the same arc. Addr is
-// the remote consumer's IP if broflake exposes it, otherwise empty.
+// the broflake mesh.
 //
-// Shape mirrors radiance/peer.ConnectionEvent so consumers (lantern-
-// core's listenPeerConnectionEvents in particular) can subscribe with
-// a single discriminator and feed both the SmC and Unbounded streams
-// into the same globe view.
+//   State     +1 on accept, -1 on close
+//   Source    consumer's IP if broflake exposes it, otherwise empty
+//   Timestamp emit time in Unix milliseconds
+//
+// Shape is identical to radiance/peer.ConnectionEvent so a single
+// subscriber can handle both the SmC-via-samizdat stream and the
+// Unbounded-via-broflake stream as one. Broflake's internal worker-
+// slot identifier is not surfaced — a consumer that needs to pair
+// accept/close events for the same arc keys off Source (or the
+// event's arrival sequence within a single connection lifetime,
+// which is what the globe currently does).
 type ConnectionEvent struct {
 	events.Event
 	State     int    `json:"state"`
-	WorkerIdx int    `json:"workerIdx"`
-	Addr      string `json:"addr"`
+	Source    string `json:"source"`
+	Timestamp int64  `json:"timestamp"`
 }
 
 var manager = &unboundedManager{}
 
 type unboundedManager struct {
-	mu      sync.Mutex
-	cancel  context.CancelFunc
-	lastCfg *C.UnboundedConfig // most recent server-supplied config
+	mu     sync.Mutex
+	cancel context.CancelFunc
+	// lastCfg + lastFeatureOn cache the server-side half of the
+	// three-condition predicate so SetEnabled can re-evaluate
+	// immediately when the local toggle flips, without waiting for
+	// the next NewConfigEvent. Both are updated atomically when a
+	// new config arrives.
+	lastCfg       *C.UnboundedConfig
+	lastFeatureOn bool
+}
+
+// shouldStart reports whether all three start conditions hold. Caller
+// must hold m.mu.
+func (m *unboundedManager) shouldStart() bool {
+	return settings.GetBool(settings.UnboundedKey) && m.lastFeatureOn && m.lastCfg != nil
 }
 
 // Enabled reports whether the local opt-in is set. Doesn't say whether
@@ -71,9 +88,10 @@ func Enabled() bool {
 }
 
 // SetEnabled flips the local opt-in. When enabling, the proxy starts
-// immediately if a server config is already cached; otherwise it
-// starts on the next config event. When disabling, the proxy stops.
-// Idempotent — calling with the current value is a no-op.
+// immediately if all three start conditions hold (local toggle + server
+// feature flag + server config cached); otherwise it stays stopped and
+// the next NewConfigEvent will reevaluate. When disabling, the proxy
+// stops. Idempotent — calling with the current value is a no-op.
 func SetEnabled(enable bool) error {
 	if Enabled() == enable {
 		return nil
@@ -82,17 +100,24 @@ func SetEnabled(enable bool) error {
 		return err
 	}
 	slog.Info("Unbounded widget proxy local opt-in changed", "enabled", enable)
-	if enable {
-		manager.mu.Lock()
-		cfg := manager.lastCfg
-		manager.mu.Unlock()
-		if cfg != nil {
-			manager.start(cfg)
-		} else {
-			slog.Info("Unbounded: enabled locally, will start when server config arrives")
-		}
-	} else {
+	if !enable {
 		manager.stop()
+		return nil
+	}
+	manager.mu.Lock()
+	shouldStart := manager.shouldStart()
+	cfg := manager.lastCfg
+	feature := manager.lastFeatureOn
+	manager.mu.Unlock()
+	if shouldStart {
+		manager.start(cfg)
+		return nil
+	}
+	switch {
+	case cfg == nil:
+		slog.Info("Unbounded: enabled locally, waiting for server config")
+	case !feature:
+		slog.Info("Unbounded: enabled locally, but server feature flag is off")
 	}
 	return nil
 }
@@ -113,13 +138,15 @@ func InitSubscription() {
 			cfg := *evt.New
 			manager.mu.Lock()
 			manager.lastCfg = cfg.Unbounded
+			manager.lastFeatureOn = cfg.Features[C.UNBOUNDED]
+			shouldRun := manager.shouldStart()
 			running := manager.cancel != nil
+			ucfg := manager.lastCfg
 			manager.mu.Unlock()
 
-			shouldRun := shouldRunUnbounded(cfg)
 			switch {
 			case shouldRun && !running:
-				manager.start(cfg.Unbounded)
+				manager.start(ucfg)
 			case !shouldRun && running:
 				manager.stop()
 			}
@@ -137,19 +164,6 @@ func Stop(_ context.Context) error {
 	return nil
 }
 
-func shouldRunUnbounded(cfg C.ConfigResponse) bool {
-	if !settings.GetBool(settings.UnboundedKey) {
-		return false
-	}
-	if !cfg.Features[C.UNBOUNDED] {
-		return false
-	}
-	if cfg.Unbounded == nil {
-		return false
-	}
-	return true
-}
-
 func (m *unboundedManager) start(ucfg *C.UnboundedConfig) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -183,11 +197,11 @@ func (m *unboundedManager) start(ucfg *C.UnboundedConfig) {
 				addrStr = addr.String()
 			}
 			slog.Debug("Unbounded: consumer connection change",
-				"state", state, "workerIdx", workerIdx, "addr", addrStr)
+				"state", state, "workerIdx", workerIdx, "source", addrStr)
 			events.Emit(ConnectionEvent{
 				State:     state,
-				WorkerIdx: workerIdx,
-				Addr:      addrStr,
+				Source:    addrStr,
+				Timestamp: time.Now().UnixMilli(),
 			})
 		}