amadan

carlos / platform Public

host: with no change index armed the telemetry pass is main's, one resolve per app (#336)

Final review finding 5. telemetrySwitchesWith over a nil watch runs
telemetrySwitchesResolve, main's loop: each (account, app) resolves both
levels on its own, so one transient account read failure leaves only that
app off. The per-pass account memo and the cross-pass last-good values apply
only when a watch is armed. TestTelemetryAccountReadOncePerPass now pins the
memo over an armed watch.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 46a738be88195e4b24210d9e76bc1746f4365476 parent 0c3e3eb
2 files changed, +92 −9
  • internal/host/readgate_test.go +47 −1
  • internal/host/telemetry_pass.go +45 −8
diff --git a/internal/host/readgate_test.go b/internal/host/readgate_test.go
index 8d3405ea8..1e3b321e6 100644
--- a/internal/host/readgate_test.go
+++ b/internal/host/readgate_test.go
@@ -319,6 +319,8 @@ func telemetryRoutes(t *testing.T, acct string, apps ...string) *registry.Regist
return reg
}
+// Armed, an account's telemetry.json is read once per pass, however many of
+// its apps the box serves.
func TestTelemetryAccountReadOncePerPass(t *testing.T) {
t.Parallel()
ctx := context.Background()
@@ -329,8 +331,10 @@ func TestTelemetryAccountReadOncePerPass(t *testing.T) {
}); err != nil {
t.Fatal(err)
}
+ w := release.NewChangeWatch(cs)
+ pollWatch(t, w)
cs.reset()
- on, err := telemetrySwitchesWith(ctx, cs, reg, newTelemetryReads(nil), newGateClock().now())
+ on, err := telemetrySwitchesWith(ctx, cs, reg, newTelemetryReads(w), newGateClock().now())
if err != nil {
t.Fatal(err)
}
@@ -342,6 +346,48 @@ func TestTelemetryAccountReadOncePerPass(t *testing.T) {
}
}
+// onceFailingGetStore fails the first Get of one key, then answers.
+type onceFailingGetStore struct {
+ *ciCountingStore
+ key string
+ failed atomic.Bool
+}
+
+func (s *onceFailingGetStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if key == s.key && s.failed.CompareAndSwap(false, true) {
+ s.note("Get", key)
+ return nil, "", errors.New("503 slow down")
+ }
+ return s.ciCountingStore.Get(ctx, key)
+}
+
+// Disarmed, the pass is main's (final review R1, finding 5): each app
+// resolves both levels on its own, so one transient account read failure
+// leaves only that app off, never the account's other apps.
+func TestTelemetryDisarmedResolvesEachAppOnItsOwn(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ cs := newCICountingStore(t)
+ reg := telemetryRoutes(t, "bab", "shop", "cafe", "deli")
+ if err := telemetry.Update(ctx, cs, telemetry.AccountKey("bab"), func(st *telemetry.Settings) {
+ st.Analytics = &telemetry.Analytics{Count: telemetry.CountEdge}
+ }); err != nil {
+ t.Fatal(err)
+ }
+ fs := &onceFailingGetStore{ciCountingStore: cs, key: telemetry.AccountKey("bab")}
+ cs.reset()
+ on, err := telemetrySwitchesWith(ctx, fs, reg, newTelemetryReads(nil), newGateClock().now())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if n := cs.count("Get", telemetry.AccountKey("bab")); n != 3 {
+ t.Fatalf("three apps read the account's telemetry.json %d time(s) disarmed, want 3 (one each, as before the index)", n)
+ }
+ if len(on) != 2 {
+ t.Fatalf("switches %v: one failed account read must leave only its own app off", on)
+ }
+}
+
func TestTelemetryGatedWithAgeBound(t *testing.T) {
t.Parallel()
ctx := context.Background()
diff --git a/internal/host/telemetry_pass.go b/internal/host/telemetry_pass.go
index 25ce64b03..c4a598e9d 100644
--- a/internal/host/telemetry_pass.go
+++ b/internal/host/telemetry_pass.go
@@ -29,8 +29,10 @@ func setAnalyticsSink(fn func(map[string]bool)) { analyticsSink.Store(&fn) }
// gate skips, or one that fails, reuses the last good value. A failed read
// marks its scope failed, so the next pass reads it again.
//
-// Disarmed, each pass gets a fresh one over a nil watch: every read is due,
-// nothing is remembered, and a failed read leaves its apps off, as before.
+// Disarmed, each pass gets a fresh one over a nil watch, and the pass is
+// main's (telemetrySwitchesResolve): every app resolves both levels on its
+// own, nothing is remembered within or across passes, and a failed read
+// leaves only that app off, as before the index.
type telemetryReads struct {
// mu is held across a whole pass, reads included: the fast converge loop
// and the instances loop both run the pass and take turns here. Nothing
@@ -75,13 +77,16 @@ func telemetrySwitches(ctx context.Context, s release.Store, reg *registry.Regis
return telemetrySwitchesWith(ctx, s, reg, currentTelemetryReads(), time.Now())
}
-// telemetrySwitchesWith is telemetrySwitches over r at now. Each account's
-// telemetry.json is read at most once per pass, however many of its apps the
-// box serves, and each app's at most once; a read is skipped when r's gate
-// says its scope has not moved and r holds a last good value. A pair with no
-// good value at either level is left OFF and logged: off is the safe side of
-// this switch.
+// telemetrySwitchesWith is telemetrySwitches over r at now. Over a watch,
+// each account's telemetry.json is read at most once per pass, however many
+// of its apps the box serves, and each app's at most once; a read is skipped
+// when r's gate says its scope has not moved and r holds a last good value.
+// A pair with no good value at either level is left OFF and logged: off is
+// the safe side of this switch. Without one (disarmed), it is main's pass.
func telemetrySwitchesWith(ctx context.Context, s release.Store, reg *registry.Registry, r *telemetryReads, now time.Time) (map[string]bool, error) {
+ if r == nil || r.w == nil {
+ return telemetrySwitchesResolve(ctx, s, reg)
+ }
routes, err := reg.List()
if err != nil {
return nil, err
@@ -125,6 +130,38 @@ func telemetrySwitchesWith(ctx context.Context, s release.Store, reg *registry.R
return on, nil
}
+// telemetrySwitchesResolve is the pass with no change index armed, exactly
+// as before the index: one Resolve (two Gets) per (account, app); a pair
+// that fails to resolve is left OFF and logged, and no other pair is.
+func telemetrySwitchesResolve(ctx context.Context, s release.Store, reg *registry.Registry) (map[string]bool, error) {
+ routes, err := reg.List()
+ if err != nil {
+ return nil, err
+ }
+ on := map[string]bool{}
+ seen := map[string]bool{}
+ for _, rt := range routes {
+ if rt.Kind == "status" || rt.App == "" {
+ continue
+ }
+ sqid := release.RouteAccount(rt)
+ key := sqid + "/" + rt.App
+ if seen[key] {
+ continue
+ }
+ seen[key] = true
+ res, err := telemetry.Resolve(ctx, s, sqid, rt.App)
+ if err != nil {
+ log.Printf("agent: telemetry resolve %s: %v (analytics stays off)", key, err)
+ continue
+ }
+ if res.Count == telemetry.CountEdge {
+ on[key] = true
+ }
+ }
+ return on, nil
+}
+
// read is key's settings for this pass: settled once per pass in pass. The
// last good value stands in when the gate says the key's scope has not moved;
// otherwise the object is read. A failed read keeps the last good value, logs