amadan

carlos / platform Public

status: fleet merge — worst-of bands, loud staleness, opt-in apps

BuildFleet merges every status/boxes/*.json snapshot into the public
fleet.json shape: per-box rows with staleness (missed 3 periods),
fleet-wide instance sums that null-propagate (any box's unknown component
nulls the summed field, never a silent partial), worst-of capacity and
replication bands, and app/service grouping that only names an app once
every one of its routes opted in via StatusNamed — one un-opted route
keeps the whole app dark. PublishFleet lands fleet.json with the same
read-etag-then-Put overwrite dance WriteBox uses, since it republishes to
the same key every pass; the embedded page-asset writes are Task 4's seam.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 85db914d7de37cbd67ecb136bed2135dc7a3600d parent d1530d7
2 files changed, +578 −0
  • internal/status/fleet.go +384 −0
  • internal/status/status_test.go +194 −0
diff --git a/internal/status/fleet.go b/internal/status/fleet.go
new file mode 100644
index 000000000..2eb750047
--- /dev/null
+++ b/internal/status/fleet.go
@@ -0,0 +1,384 @@
+package status
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "log"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// FleetApp is one opted-in app's public row — name, count, health, and the
+// release it's running. Nothing here exists for an app that hasn't set
+// StatusNamed; see BuildFleet's apps grouping.
+type FleetApp struct {
+ Name string `json:"name"`
+ Instances int `json:"instances"`
+ Healthy bool `json:"healthy"`
+ Channel string `json:"channel"`
+ Version string `json:"version"` // short sha as stored on the route
+}
+
+// FleetBox is a Box's public row inside the merged feed: everything a Box
+// carries except its own Replication (which only ever votes into the
+// fleet-wide Replication, never appears per-box) and its own Capacity
+// (same — it only ever votes into Fleet.Capacity), plus Stale, which only
+// exists at merge time since a lone Box has no "how long ago" to compare
+// against.
+type FleetBox struct {
+ Label string `json:"label"`
+ Region string `json:"region,omitempty"`
+ Load *string `json:"load"`
+ Stale bool `json:"stale"`
+ Routes RouteCounts `json:"routes"`
+}
+
+// Fleet is the merged feed: status/fleet.json. Every field here is either a
+// fleet-wide aggregate, a band word, or behind an app's own opt-in — see the
+// design spec's "rule for new fields".
+type Fleet struct {
+ GeneratedAt time.Time `json:"generated_at"`
+ Boxes []FleetBox `json:"boxes"`
+ Instances map[string]*int `json:"instances"` // total/static/resident/hibernating/awake/asleep
+ Capacity struct {
+ Level *string `json:"level"`
+ } `json:"capacity"`
+ Apps []FleetApp `json:"apps"`
+ AppsUnlisted int `json:"apps_unlisted"`
+ Services []struct {
+ Name string `json:"name"`
+ Up bool `json:"up"`
+ } `json:"services"`
+ Replication Replication `json:"replication"`
+}
+
+// capacityRank orders the capacity band words worst-last, so "worst of a
+// set of boxes" is just "the box with the highest rank" — built off
+// snapshot.go's capacityWords so the two never drift apart into disagreeing
+// about which word beats which.
+var capacityRank = map[string]int{
+ capacityWords[0]: 0,
+ capacityWords[1]: 1,
+ capacityWords[2]: 2,
+}
+
+// intPtr is the one-line "give me a pointer to this value" a lot of the
+// Instances map construction below needs — Go has no address-of-literal
+// operator, so every *int field needs a local to take the address of
+// without this helper.
+func intPtr(v int) *int { return &v }
+
+// BuildFleet merges every status/boxes/*.json (stale = GeneratedAt older
+// than 3×period), folds the local registry's routes into apps/services,
+// and never invents a number a snapshot didn't carry.
+func BuildFleet(ctx context.Context, s release.Store, routes []registry.Route, now time.Time, period time.Duration) (*Fleet, error) {
+ boxes, err := loadBoxes(ctx, s)
+ if err != nil {
+ return nil, err
+ }
+
+ f := &Fleet{GeneratedAt: now}
+
+ // stale tracks each box's staleness in lockstep with boxes — computed
+ // once here and reused by both the per-box Stale field and the
+ // worst-of/replication aggregates below, which must vote only using
+ // fresh boxes.
+ stale := make([]bool, len(boxes))
+ for i, b := range boxes {
+ stale[i] = now.Sub(b.GeneratedAt) > 3*period
+ f.Boxes = append(f.Boxes, FleetBox{
+ Label: b.Label,
+ Region: b.Region,
+ Load: b.Load,
+ Stale: stale[i],
+ Routes: b.Routes,
+ })
+ }
+
+ f.Instances = sumInstances(boxes)
+ f.Capacity.Level = worstCapacity(boxes)
+ f.Replication = worstReplication(boxes, stale)
+
+ apps, unlisted := buildApps(routes)
+ f.Apps = apps
+ f.AppsUnlisted = unlisted
+ f.Services = buildServices(routes)
+
+ return f, nil
+}
+
+// loadBoxes reads every status/boxes/*.json snapshot it can. A snapshot
+// that fails to Get or fails to unmarshal is logged and skipped — one box's
+// bad write must never take the whole fleet page down, per the plan's
+// "corrupt snapshot = logged skip, never fatal" rule.
+func loadBoxes(ctx context.Context, s release.Store) ([]Box, error) {
+ keys, err := s.List(ctx, "status/boxes/")
+ if err != nil {
+ return nil, err
+ }
+ var boxes []Box
+ for _, key := range keys {
+ body, _, err := s.Get(ctx, key)
+ if err != nil {
+ log.Printf("status: fleet: skipping %s: read failed: %v", key, err)
+ continue
+ }
+ var b Box
+ if err := json.Unmarshal(body, &b); err != nil {
+ log.Printf("status: fleet: skipping %s: corrupt snapshot: %v", key, err)
+ continue
+ }
+ boxes = append(boxes, b)
+ }
+ return boxes, nil
+}
+
+// sumInstances adds each RouteCounts component across every box — stale
+// boxes included, since staleness is a "trust this less" signal rendered
+// loudly per-box, not a reason to silently drop a box's last-known counts
+// from the fleet total. Static/Resident/Hibernating are plain ints on Box
+// (always known), so their sums are always known too. Awake/Asleep are
+// *int on Box (null when a box's socket table was unreadable), so per the
+// plan's null-propagation rule, ANY box contributing null nulls the whole
+// summed field — a partial sum would read as more certain than it is.
+func sumInstances(boxes []Box) map[string]*int {
+ var totalStatic, totalResident, totalHibernating, awakeSum, asleepSum int
+ awakeKnown, asleepKnown := true, true
+ for _, b := range boxes {
+ totalStatic += b.Routes.Static
+ totalResident += b.Routes.Resident
+ totalHibernating += b.Routes.Hibernating
+ if b.Routes.Awake == nil {
+ awakeKnown = false
+ } else {
+ awakeSum += *b.Routes.Awake
+ }
+ if b.Routes.Asleep == nil {
+ asleepKnown = false
+ } else {
+ asleepSum += *b.Routes.Asleep
+ }
+ }
+ instances := map[string]*int{
+ "total": intPtr(totalStatic + totalResident + totalHibernating),
+ "static": intPtr(totalStatic),
+ "resident": intPtr(totalResident),
+ "hibernating": intPtr(totalHibernating),
+ }
+ if awakeKnown {
+ instances["awake"] = intPtr(awakeSum)
+ } else {
+ instances["awake"] = nil
+ }
+ if asleepKnown {
+ instances["asleep"] = intPtr(asleepSum)
+ } else {
+ instances["asleep"] = nil
+ }
+ return instances
+}
+
+// worstCapacity picks the worst capacity band among ALL boxes, stale ones
+// included: a box that stopped reporting still had a last-known fill level,
+// and dropping it from the vote just because its snapshot aged out would
+// hide a real "getting full" or "at capacity" reading behind the newer,
+// possibly-rosier boxes — exactly the silent-omission failure the plan's
+// privacy rules refuse to allow, just pointed at capacity instead of a
+// hostname. (Contrast worstReplication below, which the plan explicitly
+// scopes to fresh boxes — staleness there means "we don't know", not "we
+// know it was bad".) A box with an unknown (nil) band doesn't vote — nil
+// never invents a word. All-nil input yields nil.
+func worstCapacity(boxes []Box) *string {
+ var worst *string
+ worstRank := -1
+ for _, b := range boxes {
+ if b.Capacity == nil {
+ continue
+ }
+ rank, ok := capacityRank[*b.Capacity]
+ if !ok {
+ continue // defensive: an unrecognized band word doesn't vote
+ }
+ if rank > worstRank {
+ worstRank = rank
+ v := *b.Capacity
+ worst = &v
+ }
+ }
+ return worst
+}
+
+// worstReplication picks the worst replication reading among non-stale
+// boxes: unhealthy beats healthy beats unknown, and ties within a band
+// break toward the larger lag (still the worse reading). Nothing fresh
+// with a known reading yields the zero Replication (both fields nil) —
+// unknown, not a guessed "healthy".
+func worstReplication(boxes []Box, stale []bool) Replication {
+ var worst Replication
+ haveWorst := false
+ for i, b := range boxes {
+ if stale[i] || b.Replication.Healthy == nil {
+ continue
+ }
+ switch {
+ case !haveWorst:
+ worst = b.Replication
+ haveWorst = true
+ case !*b.Replication.Healthy && *worst.Healthy:
+ // This box is unhealthy and the current worst isn't — a real
+ // outage always outranks "just some lag".
+ worst = b.Replication
+ case *b.Replication.Healthy == *worst.Healthy &&
+ b.Replication.LagSecs != nil && worst.LagSecs != nil &&
+ *b.Replication.LagSecs > *worst.LagSecs:
+ // Same health band; the larger lag is the worse reading.
+ worst = b.Replication
+ }
+ }
+ return worst
+}
+
+// routeGroup collects one App name's routes in the order they appeared in
+// the input slice — used for both the apps and services groupings below,
+// which differ only in which Kind they filter on and how they reduce a
+// group down to one published row.
+type routeGroup struct {
+ name string
+ routes []registry.Route
+}
+
+// groupRoutesByApp buckets routes matching kind into per-App groups,
+// preserving first-seen order so output is deterministic without needing a
+// separate sort pass.
+func groupRoutesByApp(routes []registry.Route, kind string) []routeGroup {
+ index := map[string]int{}
+ var groups []routeGroup
+ for _, r := range routes {
+ if r.Kind != kind {
+ continue
+ }
+ i, ok := index[r.App]
+ if !ok {
+ i = len(groups)
+ index[r.App] = i
+ groups = append(groups, routeGroup{name: r.App})
+ }
+ groups[i].routes = append(groups[i].routes, r)
+ }
+ return groups
+}
+
+// buildApps groups the registry's "instance" routes by App and publishes
+// only groups where EVERY route opted in via StatusNamed — one un-opted
+// route keeps its whole app dark, never a partial reveal. Everything else
+// folds into the unlisted count so the aggregate is still honest about how
+// many apps exist without naming the ones that didn't ask to be named.
+func buildApps(routes []registry.Route) ([]FleetApp, int) {
+ groups := groupRoutesByApp(routes, "instance")
+
+ var apps []FleetApp
+ for _, g := range groups {
+ allNamed := true
+ healthy := true
+ for _, r := range g.routes {
+ if !r.StatusNamed {
+ allNamed = false
+ }
+ if r.Status != "ready" {
+ healthy = false
+ }
+ }
+ if !allNamed {
+ continue
+ }
+ first := g.routes[0]
+ for _, r := range g.routes[1:] {
+ if r.Channel != first.Channel || r.Version != first.Version {
+ log.Printf("status: fleet: app %q routes disagree on channel/version (using first route's %q/%q)",
+ g.name, first.Channel, first.Version)
+ break
+ }
+ }
+ apps = append(apps, FleetApp{
+ Name: g.name,
+ Instances: len(g.routes),
+ Healthy: healthy,
+ Channel: first.Channel,
+ Version: first.Version,
+ })
+ }
+ return apps, len(groups) - len(apps)
+}
+
+// buildServices groups the registry's "service" routes by App and reports
+// each as up only if every route in the group is ready. Service names are
+// platform-reserved (svc/*), not tenant hostnames, so — unlike apps —
+// there's no opt-in gate here.
+func buildServices(routes []registry.Route) []struct {
+ Name string `json:"name"`
+ Up bool `json:"up"`
+} {
+ groups := groupRoutesByApp(routes, "service")
+
+ var services []struct {
+ Name string `json:"name"`
+ Up bool `json:"up"`
+ }
+ for _, g := range groups {
+ up := true
+ for _, r := range g.routes {
+ if r.Status != "ready" {
+ up = false
+ }
+ }
+ services = append(services, struct {
+ Name string `json:"name"`
+ Up bool `json:"up"`
+ }{Name: g.name, Up: up})
+ }
+ return services
+}
+
+// PublishFleet writes fleet.json to status/. Page assets (index.html, the
+// stylesheet) are embedded and published alongside it starting in Task 4 —
+// this function's single Put is the seam that loop lands in; it isn't a
+// TODO so much as the honest scope of what this task builds.
+func PublishFleet(ctx context.Context, s release.Store, f *Fleet) error {
+ body, err := json.MarshalIndent(f, "", " ")
+ if err != nil {
+ return err
+ }
+ return putOverwrite(ctx, s, "status/fleet.json", body)
+}
+
+// putOverwrite lands body at key with last-writer-wins semantics — the same
+// read-etag-then-Put dance WriteBox uses, for the same reason: every
+// status/ object PublishFleet writes gets republished on the same key each
+// pass (fleet.json every 5 minutes, alongside the boxes), and
+// release.Store's Put is CAS-only, so a bare create-only Put would fail
+// every tick after the first with ErrCASConflict. One retry on a lost race;
+// a second conflict is surfaced rather than retried again, since the status
+// lease is supposed to make PublishFleet the only writer touching these
+// keys — two losses in a row past the lease means something else is wrong.
+func putOverwrite(ctx context.Context, s release.Store, key string, body []byte) error {
+ etag, err := currentEtag(ctx, s, key)
+ if err != nil {
+ return err
+ }
+ if _, err := s.Put(ctx, key, body, etag); err != nil {
+ if !errors.Is(err, release.ErrCASConflict) {
+ return err
+ }
+ etag, err = currentEtag(ctx, s, key)
+ if err != nil {
+ return err
+ }
+ if _, err := s.Put(ctx, key, body, etag); err != nil {
+ return err
+ }
+ }
+ return nil
+}
diff --git a/internal/status/status_test.go b/internal/status/status_test.go
index ca8195b51..163f2de8b 100644
--- a/internal/status/status_test.go
+++ b/internal/status/status_test.go
@@ -189,6 +189,200 @@ func TestWriteBoxRetriesOnceOnCASConflict(t *testing.T) {
}
}
+// ---- Task 3: fleet merge + publisher --------------------------------------
+
+// teststore is a fresh DirStore over a scratch directory — every fleet test
+// gets an isolated bucket so writes in one test can never leak state into
+// another.
+func teststore(t *testing.T) release.Store {
+ t.Helper()
+ return release.NewDirStore(t.TempDir())
+}
+
+// boxFixture builds a minimal Box for tests that only care about label,
+// capacity band, and generation time — the three things BuildFleet's
+// worst-of/staleness logic actually looks at.
+func boxFixture(label, capacity string, generatedAt time.Time) status.Box {
+ c := capacity
+ return status.Box{
+ GeneratedAt: generatedAt,
+ Label: label,
+ Routes: status.RouteCounts{},
+ Capacity: &c,
+ }
+}
+
+func TestFleetAppsOptIn(t *testing.T) {
+ routes := []registry.Route{
+ {Host: "kass.training", Kind: "instance", App: "kass", Status: "ready",
+ Channel: "stable", Version: "aedcae3", StatusNamed: true},
+ {Host: "hidden.app", Kind: "instance", App: "hidden", Status: "ready"},
+ {Host: "svc.pdf.internal", Kind: "service", App: "svc/pdf", Status: "ready"},
+ }
+ s := teststore(t) // DirStore over t.TempDir(); write one Box via WriteBox first
+ if err := status.WriteBox(context.Background(), s, boxFixture("solo", "plenty", time.Now())); err != nil {
+ t.Fatal(err)
+ }
+ f, err := status.BuildFleet(context.Background(), s, routes, time.Now(), 5*time.Minute)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(f.Apps) != 1 || f.Apps[0].Name != "kass" || f.Apps[0].Version != "aedcae3" {
+ t.Errorf("apps = %+v, want exactly the opted-in kass", f.Apps)
+ }
+ if f.AppsUnlisted != 1 { // hidden; svc/* routes are services, not apps
+ t.Errorf("apps_unlisted = %d, want 1", f.AppsUnlisted)
+ }
+ if len(f.Services) != 1 || f.Services[0].Name != "svc/pdf" || !f.Services[0].Up {
+ t.Errorf("services = %+v", f.Services)
+ }
+ j, _ := json.Marshal(f)
+ if strings.Contains(string(j), "hidden") {
+ t.Errorf("un-opted app leaked: %s", j)
+ }
+}
+
+func TestFleetWorstOfAndStale(t *testing.T) {
+ s := teststore(t)
+ now := time.Now()
+ old, fresh := boxFixture("old", "at-capacity", now.Add(-16*time.Minute)),
+ boxFixture("fresh", "plenty", now) // helper: Box with that capacity word
+ for _, b := range []status.Box{old, fresh} {
+ if err := status.WriteBox(context.Background(), s, b); err != nil {
+ t.Fatal(err)
+ }
+ }
+ f, _ := status.BuildFleet(context.Background(), s, nil, now, 5*time.Minute)
+ if f.Capacity.Level == nil || *f.Capacity.Level != "at-capacity" {
+ t.Errorf("capacity must be the worst box's band, got %v", f.Capacity.Level)
+ }
+ var st, fr *status.FleetBox
+ for i := range f.Boxes {
+ if f.Boxes[i].Label == "old" {
+ st = &f.Boxes[i]
+ } else {
+ fr = &f.Boxes[i]
+ }
+ }
+ if !st.Stale || fr.Stale {
+ t.Error("boxes older than 3 periods must mark stale — loudly, never dropped")
+ }
+}
+
+// TestFleetInstanceSumsNullPropagate proves the "unknowns are null, never 0"
+// rule holds through aggregation too: one box with an unreadable socket
+// table (Awake/Asleep null) must null the FLEET-WIDE awake/asleep sum, not
+// silently drop that box's contribution and report a partial — and
+// misleadingly confident — total.
+func TestFleetInstanceSumsNullPropagate(t *testing.T) {
+ s := teststore(t)
+ now := time.Now()
+ known := status.CollectBox(
+ []registry.Route{{Host: "a", Kind: "instance", Hibernate: true, Socket: "/a.sock"}},
+ status.Probes{MachineID: "known", Listening: func(string) *bool { v := true; return &v }},
+ now,
+ )
+ unknown := status.CollectBox(
+ []registry.Route{{Host: "b", Kind: "instance", Hibernate: true, Socket: "/b.sock"}},
+ status.Probes{MachineID: "unknown"}, // nil Listening -> Awake/Asleep null
+ now,
+ )
+ for _, b := range []status.Box{known, unknown} {
+ if err := status.WriteBox(context.Background(), s, b); err != nil {
+ t.Fatal(err)
+ }
+ }
+ f, err := status.BuildFleet(context.Background(), s, nil, now, 5*time.Minute)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if f.Instances["awake"] != nil {
+ t.Errorf("fleet awake = %v, want nil (one box's socket table was unreadable)", *f.Instances["awake"])
+ }
+ if f.Instances["static"] == nil || *f.Instances["static"] != 0 {
+ t.Errorf("fleet static = %v, want 0 (a known, non-null component)", f.Instances["static"])
+ }
+ if f.Instances["hibernating"] == nil || *f.Instances["hibernating"] != 2 {
+ t.Errorf("fleet hibernating = %v, want 2", f.Instances["hibernating"])
+ }
+}
+
+// TestFleetEmptyIsValid locks down the zero-snapshot state: BuildFleet must
+// not error when status/boxes/ is empty, and must report zero counts (not
+// null — there's nothing unknown about an empty fleet) with a null capacity
+// band (there's no box to have voted on one). This is the console's
+// "no snapshots yet" rendering.
+func TestFleetEmptyIsValid(t *testing.T) {
+ s := teststore(t)
+ f, err := status.BuildFleet(context.Background(), s, nil, time.Now(), 5*time.Minute)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(f.Boxes) != 0 {
+ t.Errorf("boxes = %+v, want none", f.Boxes)
+ }
+ if f.Capacity.Level != nil {
+ t.Errorf("capacity.level = %v, want nil with no boxes to vote", *f.Capacity.Level)
+ }
+ if f.Instances["total"] == nil || *f.Instances["total"] != 0 {
+ t.Errorf("instances.total = %v, want 0", f.Instances["total"])
+ }
+ if len(f.Apps) != 0 || f.AppsUnlisted != 0 || len(f.Services) != 0 {
+ t.Errorf("apps/services must be empty with no routes, got apps=%+v unlisted=%d services=%+v", f.Apps, f.AppsUnlisted, f.Services)
+ }
+}
+
+// TestFleetSkipsCorruptSnapshot proves a corrupt status/boxes/*.json is a
+// logged skip, never a fatal error — one box's bad write must not take the
+// whole fleet page down.
+func TestFleetSkipsCorruptSnapshot(t *testing.T) {
+ s := teststore(t)
+ now := time.Now()
+ if err := status.WriteBox(context.Background(), s, boxFixture("good", "plenty", now)); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := s.Put(context.Background(), "status/boxes/broken.json", []byte("{not json"), ""); err != nil {
+ t.Fatal(err)
+ }
+ f, err := status.BuildFleet(context.Background(), s, nil, now, 5*time.Minute)
+ if err != nil {
+ t.Fatalf("BuildFleet must not fail on a corrupt snapshot: %v", err)
+ }
+ if len(f.Boxes) != 1 || f.Boxes[0].Label != "good" {
+ t.Errorf("boxes = %+v, want only the good one (corrupt skipped)", f.Boxes)
+ }
+}
+
+// TestPublishFleetWritesAndOverwrites mirrors WriteBox's own overwrite test:
+// PublishFleet lands status/fleet.json on every publish pass (every 5
+// minutes, per the plan), so a second call must overwrite cleanly rather
+// than failing with ErrCASConflict against release.Store's create-only Put.
+func TestPublishFleetWritesAndOverwrites(t *testing.T) {
+ s := teststore(t)
+ ctx := context.Background()
+
+ first := &status.Fleet{GeneratedAt: time.Unix(1000, 0)}
+ if err := status.PublishFleet(ctx, s, first); err != nil {
+ t.Fatalf("first PublishFleet: %v", err)
+ }
+ second := &status.Fleet{GeneratedAt: time.Unix(2000, 0)}
+ if err := status.PublishFleet(ctx, s, second); err != nil {
+ t.Fatalf("second PublishFleet (must overwrite, not conflict): %v", err)
+ }
+
+ got, _, err := s.Get(ctx, "status/fleet.json")
+ if err != nil {
+ t.Fatalf("Get status/fleet.json: %v", err)
+ }
+ var round status.Fleet
+ if err := json.Unmarshal(got, &round); err != nil {
+ t.Fatalf("unmarshal published fleet: %v", err)
+ }
+ if !round.GeneratedAt.Equal(second.GeneratedAt) {
+ t.Errorf("stored fleet has GeneratedAt %v, want second publish's %v", round.GeneratedAt, second.GeneratedAt)
+ }
+}
+
// TestWriteBoxReturnsErrorOnDoubleCASConflict proves WriteBox does NOT retry
// a second time: if the retried Put also loses the race, that error must
// reach the caller rather than being swallowed or retried forever.