feat: keep the service's hours for 90 days and serve them to the stats Worker

CLAUDE.md

@@ -42,8 +42,8 @@ maintains it.
   Memory holds live readings, and only while the roster's
   grant serves a copy; SQLite holds only the home id, the box key, the signed
   roster, owners' suspensions, tombstones, APNs registrations, the hashes of
-  extension tokens when each away box's alert is due, browser notice sent markers and
-  anonymous daily fleet and energy totals. Browser endpoints and box-encrypted offline
+  extension tokens when each away box's alert is due, browser notice sent markers,
+  anonymous daily fleet and energy totals, and the service's own hourly counts. Browser endpoints and box-encrypted offline
   notices live only in the signed roster. Nothing logs
   addresses.
 - **A phone can only narrow, or talk to its own box.** On `/v1/app` it can
@@ -70,6 +70,10 @@ maintains it.
   one, because other parties' clients follow no redirect.
 - **Counts name nobody.** `/metrics` listens on loopback only, and it and
   the minute's log line sum over all homes: no home id, device or address.
+  The registry keeps the same sums an hour at a time for 90 days, and
+  `/v1/ops` serves them, with the closed fleet and energy day totals, only to
+  the stats Worker's key. A label is a word fixed in the code
+  (`TestCountsNameNobody`); never add one that could hold a home.
 - **A deploy drains.** On SIGTERM every socket, handshakes included, hears
   4503 with a spread before the process exits, and no drained box starts
   `box.unreachable`. A part that holds sockets is a `drainer` in

README.md

@@ -26,8 +26,8 @@ to it yet.
 - What the service holds follows the newest roster's grant, never `hello`:
   the readable copy can turn on or off while the uplink stays up.
 - The registry (SQLite) holds the home id, the box key, the signed roster,
-  owners' suspensions, APNs registrations, the hashes of extension tokens
-  and when an away box's alert is due. A box whose owner turns the readable
+  owners' suspensions, APNs registrations, the hashes of extension tokens,
+  when an away box's alert is due, and the service's own hourly counts. A box whose owner turns the readable
   copy off gets that copy deleted and a signed receipt (`purged {what:
   "copy"}`); a box whose link ends gets its home deleted and a receipt.
 - **Room to grow.** Each home lives on one shard; an instance that does not
@@ -36,7 +36,8 @@ to it yet.
   one instance holds every home. A deploy closes every socket with 4503 and
   a spread, so boxes and phones come back over seconds rather than at once.
   `/metrics` on loopback and one log line a minute count what the service
-  does without naming a home. `deploy/README.md` has the load numbers and
+  does without naming a home; the registry keeps the same counts an hour
+  at a time for 90 days ([Internal operations](#internal-operations)). `deploy/README.md` has the load numbers and
   when to add a shard.
 - **A restart forgets no one.** At start the service reads every signed
   roster back from the registry and checks it with the home's key, so a
@@ -585,3 +586,48 @@ OHTTP should be outside GDPR, but reading data from the owner's equipment to
 make them may still need consent under the ePrivacy rule (Swedish LEK,
 chapter 9, section 28). If it does, neither the opt-out for boxes without
 Cloud nor the requirement with Cloud holds.
+
+## Internal operations
+
+Sourceful runs the service and the fleet from one internal view: the private
+page of stats.ftw.energy, behind Cloudflare Access. It needs no system of
+its own. The service keeps its counts; the stats Worker pulls them once an
+hour into its D1 database.
+
+**What the service keeps.** Every hour `internal/metrics` hands the hour to
+the registry (`ops_hours`, one row an hour): how much each counter rose,
+each gauge's low, mean and high over the minutes sampled, each latency's
+buckets, how much of the hour a process ran, and how many processes started
+in it. A deploy hands over its part of the hour after the drain, and the
+next process adds the rest to the same row. The series are the ones
+`/metrics` serves: sockets by kind and by transport (QUIC or WebSocket),
+homes held and connected, bytes written to boxes and phones, tun frames and
+bytes, drops by reason, push queues and the answers of Apple and of browsers'
+push services, MCP requests and tool calls by outcome and their time,
+handshake and forward latency, the drain's sockets and the Go runtime.
+Hours go after 90 days, at each write and at upkeep.
+
+**What it serves.** `GET /v1/ops` answers only the key the Worker holds,
+`Authorization: Bearer <key>`. The service holds the key's SHA-256
+(`-ops-key-sha256`), which is no secret; without it the route answers 404
+to everyone, and the hours are kept all the same. An answer carries at most
+a week of settled hours (`?since=`, five minutes after an hour ends), and the
+exact totals of the anonymous fleet and energy reports for closed days: a
+fleet day once it has ended, an energy date once no report may name it, as
+`/v1/energy/public` has it. Below the public group size of 5 too, but never
+an open day, so two answers never differ by one report. `/v1/ops` is
+`ftw.cloud-ops/1`; the Worker refuses any field it does not name.
+
+**Why none of it is about a household.** Each series sums over the whole
+process. A label is a word fixed in the code, never a home, a device, a key
+or an address (`TestCountsNameNobody`), and no series is made for a home
+(`TestAnHourOfCountsNamesNobody`). The fleet and energy days are totals of
+reports that carry no id and keep nothing per report. The limits, plainly:
+while the fleet is a handful of homes, an hour's phone count or a closed
+day's single report says something about those few, and staff who know
+them could guess. The feed gives nobody more than the totals, and nobody
+outside Access sees them. [privacy/internal-operations.md](privacy/internal-operations.md)
+has the argument and the retention in full.
+
+Per-home data stays where it is: in memory for the home's own phones, and in
+support access only when an owner opts in for a ticket.

cmd/ftwcloudd/main.go

@@ -44,6 +44,7 @@ import (
 	"github.com/srcfl/ftw-cloud/internal/mcp"
 	"github.com/srcfl/ftw-cloud/internal/metrics"
 	"github.com/srcfl/ftw-cloud/internal/oauth"
+	"github.com/srcfl/ftw-cloud/internal/ops"
 	"github.com/srcfl/ftw-cloud/internal/push"
 	"github.com/srcfl/ftw-cloud/internal/quicdoor"
 	"github.com/srcfl/ftw-cloud/internal/registry"
@@ -104,6 +105,7 @@ func serve(args []string) error {
 	shardsPath := fs.String("shards", "", "shard directory, JSON; without it this instance holds every home")
 	statsPolicy := fs.String("statistics-policy", "", "approved statistics policy; empty keeps hourly collection off pending C8 sign-off")
 	metricsAddr := fs.String("metrics-listen", defaultMetricsAddr, "loopback address for /metrics; empty turns it off")
+	opsKey := fs.String("ops-key-sha256", "", "hex SHA-256 of the key the project stats Worker reads /v1/ops with; empty serves no one")
 	restartRate := fs.Int("restart-rate", defaultRestartRate, "sockets a second the next instance can take after a restart; sets the spread of the redial")
 	restartMin := fs.Duration("restart-spread-min", 15*time.Second, "the shortest spread: longer than the port stays closed while the next container starts")
 	h3 := fs.Bool("h3", true, "also serve HTTP/3 on the same address over UDP, and say so in Alt-Svc")
@@ -179,9 +181,17 @@ func serve(args []string) error {
 	if err != nil {
 		return err
 	}
+	// The service's own hours, kept 90 days for Sourceful's internal view.
+	opsSvc, err := ops.New(reg, shards.Self())
+	if err != nil {
+		return err
+	}
+	if err := opsSvc.UseKey(*opsKey); err != nil {
+		log.Error("ops-key-sha256 is not 64 hex digits; /v1/ops stays off, and the hours are kept", "err", err)
+	}
 	tokens := token.New(st, reg, log)
 	tokens.Shards = shards
-	srv := &server.Server{Store: st, Registry: reg, Uplink: up, Tokens: tokens, Fleet: fleetSvc, Gateway: gatewaySvc, Statistics: usageSvc, Log: log}
+	srv := &server.Server{Store: st, Registry: reg, Uplink: up, Tokens: tokens, Fleet: fleetSvc, Gateway: gatewaySvc, Statistics: usageSvc, Ops: opsSvc, Log: log}
 
 	// Assistants: the authorization server, with agent keys in ftwkeysd.
 	agentKeys := keys.Dial(*keysSocket)
@@ -348,6 +358,10 @@ func serve(args []string) error {
 		}
 	}
 	go metrics.Default.Summarise(ctx, os.Stdout, shards.Self(), time.Minute)
+	// The same series, an hour at a time, into the registry. A deploy hands
+	// over the hour so far after the drain, so the drain is in it.
+	keeper := metrics.Default.NewKeeper(opsSvc, func(err error) { log.Warn("an hour of metrics was not kept", "err", err) })
+	go keeper.Run(ctx)
 
 	// A deploy: stop accepting, tell every box and phone to come back within
 	// a spread the next instance can take, let the queued alerts go, then
@@ -382,9 +396,13 @@ func serve(args []string) error {
 			left += d.WaitClosed(time.Until(deadline))
 		}
 		log.Info("drained for a restart", "sockets", sockets, "still_open", left, "spread", spread)
+		drains.Inc()
+		drainedSockets.Add(int64(sockets))
+		drainLeftOpen.Add(int64(left))
 		if left := pushSvc.Flush(3 * time.Second); left > 0 {
 			log.Warn("alerts queued or waiting to go again at exit; they are lost", "count", left)
 		}
+		keeper.Close()
 		stopPush()
 	}()
 	if door != nil {
@@ -425,6 +443,13 @@ type drainer interface {
 // drainWait bounds the wait for the clients' answers to the drain's close.
 const drainWait = 2 * time.Second
 
+// What a deploy's drain did, kept in the hour it ends.
+var (
+	drains         = metrics.Default.Counter("ftwcloud_drains_total", "Drains for a restart.")
+	drainedSockets = metrics.Default.Counter("ftwcloud_drained_sockets_total", "Sockets a drain closed with 4503, handshakes included.")
+	drainLeftOpen  = metrics.Default.Counter("ftwcloud_drain_left_open_total", "Sockets whose clients had not answered the drain's close when it stopped waiting.")
+)
+
 // defaultRestartRate is how many sockets a second one instance takes on
 // the host, with room to spare: deploy/README.md, "Load", has the numbers.
 const defaultRestartRate = 500

cmd/ftwcloudd/main_test.go

@@ -1,10 +1,43 @@
 package main
 
 import (
+	"regexp"
+	"slices"
+	"strings"
 	"testing"
 	"time"
+
+	"github.com/srcfl/ftw-cloud/internal/metrics"
 )
 
+// Every series the service keeps, which /v1/ops keeps for 90 days, carries
+// only labels from this list, each a word fixed in the code. A label that
+// could hold a home, a device or an address is a change to argue for here.
+func TestCountsNameNobody(t *testing.T) {
+	var b strings.Builder
+	if _, err := metrics.Default.WriteTo(&b); err != nil {
+		t.Fatal(err)
+	}
+	pair := regexp.MustCompile(`([a-z_]+)="([^"]*)"`)
+	word := regexp.MustCompile(`^[a-z0-9_]{1,40}$|^[0-9.e+-]+$|^\+Inf$`)
+	allowed := []string{"kind", "reason", "path", "dir", "result", "transport", "outcome", "le"}
+	n := 0
+	for _, line := range strings.Split(b.String(), "\n") {
+		if strings.HasPrefix(line, "#") || line == "" {
+			continue
+		}
+		n++
+		for _, m := range pair.FindAllStringSubmatch(line, -1) {
+			if !slices.Contains(allowed, m[1]) || !word.MatchString(m[2]) {
+				t.Fatalf("label %s=%q in %s", m[1], m[2], line)
+			}
+		}
+	}
+	if n < 50 {
+		t.Fatalf("only %d series: is every package's set registered?", n)
+	}
+}
+
 func TestTheRestartSpreadFollowsTheSockets(t *testing.T) {
 	for _, c := range []struct {
 		sockets, rate int

deploy/README.md

@@ -215,6 +215,23 @@ closed port, so the spread's floor can drop to a few seconds.
   Logs (the `awslogs` log driver, and `logs:CreateLogStream` and
   `logs:PutLogEvents` for the instance role, which the owner adds), each
   field is a metric in the `FTWCloud` namespace, by shard.
+- **The hours, for 90 days.** The registry keeps the same counts an hour
+  at a time, and the private page of stats.ftw.energy reads them from
+  `/v1/ops` once an hour ([README](../README.md#internal-operations)). To
+  turn the feed on, make a key and give it to both ends:
+
+  ```bash
+  key=$(openssl rand -base64 32 | tr '+/' '-_' | tr -d '=')
+  printf %s "$key" | npx wrangler secret put CLOUD_OPS_KEY --config stats/wrangler.jsonc   # in srcfl/ftw-webapp
+  printf %s "$key" | shasum -a 256 | cut -d' ' -f1                                         # FTW_CLOUD_OPS_KEY_SHA256
+  ```
+
+  Put the hash in `/etc/ftw-cloud/compose.env` as
+  `FTW_CLOUD_OPS_KEY_SHA256=`; the next deploy serves the feed. The hash is
+  no secret, the key is: it lives only in the Worker. Until the service has
+  a hash the Worker logs `awaiting key on Cloud`; while the two differ, a
+  401. A hash that is not 64 hex digits is logged at start and leaves the
+  feed off; the service runs. To rotate, do all three again.
 - What to watch: `forward_seconds_*_p99` (time inside the service),
   `drops_*` (any rise), `push_queued_alert`, `homes_connected`,
   `go_memory_bytes`.

deploy/compose.yml

@@ -30,6 +30,9 @@ services:
       - -noise-key=/run/secrets/noise/noise.key
       - -ohttp-key=/run/secrets/ohttp/ohttp.key
       - -keys-socket=/run/ftw-cloud-keys/keys.sock
+      # The SHA-256 of the key the stats Worker reads /v1/ops with: no
+      # secret. Empty turns the feed off; the hours are kept all the same.
+      - -ops-key-sha256=${FTW_CLOUD_OPS_KEY_SHA256:-}
     environment:
       # /source publishes it, so anyone can compare it with what CI built.
       FTW_CLOUD_IMAGE: ${FTW_CLOUD_IMAGE}

internal/appsession/appsession.go

@@ -94,6 +94,8 @@ type Exchanger interface {
 // What the app sessions count, summed over all homes.
 var (
 	appSockets     = metrics.Sockets("app")
+	appBytesOut    = metrics.BytesOut("app")
+	appOver        = map[string]*metrics.Gauge{cloudwire.TransportQUIC: metrics.TransportSockets("app", cloudwire.TransportQUIC), cloudwire.TransportWebSocket: metrics.TransportSockets("app", cloudwire.TransportWebSocket)}
 	appHandshake   = metrics.Handshake("app")
 	onceHandshake  = metrics.Handshake("once")
 	envelopeOut    = metrics.Forward("envelope")
@@ -221,6 +223,10 @@ func (s *Server) ServeConn(conn wire.Conn, home string) {
 	defer conn.Close()
 	appSockets.Add(1)
 	defer appSockets.Add(-1)
+	if over := appOver[wire.Transport(conn)]; over != nil {
+		over.Add(1)
+		defer over.Add(-1)
+	}
 	if !cloudwire.ValidHome(home) {
 		_ = conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(cloudwire.CloseProtocol, ""), time.Now().Add(time.Second))
 		return
@@ -538,7 +544,11 @@ func (c *phoneConn) writeAs(kind int, msg []byte) error {
 	if err := c.conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil {
 		return err
 	}
-	return c.conn.WriteMessage(wire.Kind(c.conn, kind), msg)
+	if err := c.conn.WriteMessage(wire.Kind(c.conn, kind), msg); err != nil {
+		return err
+	}
+	appBytesOut.Add(int64(len(msg)))
+	return nil
 }
 
 // forward sends one box envelope as the box sent it. An envelope too large

internal/mcp/mcp.go

@@ -32,9 +32,22 @@ import (
 	"sync"
 	"time"
 
+	"github.com/srcfl/ftw-cloud/internal/metrics"
 	"github.com/srcfl/ftw-cloud/internal/oauth"
 )
 
+// What the service counts of assistants, summed over every home and agent.
+var (
+	mcpOutcomes = map[string]*metrics.Counter{}
+	mcpCall     = metrics.Default.Histogram("ftwcloud_mcp_call_seconds", "From a tool's start to its result, the box's answer included.")
+)
+
+func init() {
+	for _, o := range []string{"refused", "rate_limited", "ok", "tool_error", "failed", "unknown_tool", "unsupported", "step_up", "busy"} {
+		mcpOutcomes[o] = metrics.Default.Counter("ftwcloud_mcp_total", "MCP requests refused before a call, and tool calls, by outcome.", "outcome", o)
+	}
+}
+
 // Versions.
 const (
 	// Modern is the revision this server is written to.
@@ -196,10 +209,12 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	_ = http.NewResponseController(w).SetReadDeadline(time.Now().Add(10 * time.Second))
 	grant, aerr := s.auth.Authenticate(r)
 	if aerr != nil {
+		mcpOutcomes["refused"].Inc()
 		oauth.Challenge(w, profileAuthError(r, aerr))
 		return
 	}
 	if !s.limit.allow(grant.Subject.Home+"/"+grant.Subject.Agent, s.now()) {
+		mcpOutcomes["rate_limited"].Inc()
 		w.Header().Set("Retry-After", "1")
 		writeRPCError(w, http.StatusTooManyRequests, nil, codeInternal, "too many requests; slow down", nil)
 		return
@@ -375,6 +390,7 @@ type callParams struct {
 
 func (s *Server) call(w http.ResponseWriter, r *http.Request, id json.RawMessage, p callParams, g *oauth.Grant, modern bool) {
 	if unsupportedDiagnostic(p.Name) {
+		mcpOutcomes["unsupported"].Inc()
 		result := toolError("This connector does not provide this operation (E_UNSUPPORTED). Use measurement_diagnostics, health_get or logs_tail.")
 		result["structuredContent"] = map[string]any{"source": "service", "data": map[string]any{"available": false, "code": "E_UNSUPPORTED", "operation": p.Name, "alternatives": []string{"measurement_diagnostics", "health_get", "logs_tail"}}}
 		writeResult(w, id, modern, result)
@@ -385,6 +401,7 @@ func (s *Server) call(w http.ResponseWriter, r *http.Request, id json.RawMessage
 		t = nil
 	}
 	if t == nil {
+		mcpOutcomes["unknown_tool"].Inc()
 		status := http.StatusOK
 		if modern {
 			status = http.StatusBadRequest
@@ -395,6 +412,7 @@ func (s *Server) call(w http.ResponseWriter, r *http.Request, id json.RawMessage
 	required := t.requiredScopes()
 	if slices.ContainsFunc(required, func(scope string) bool { return !g.Has(scope) }) {
 		// Step up: the client asks the owner for the right, then retries.
+		mcpOutcomes["step_up"].Inc()
 		scopes := append(slices.Clone(g.Scopes), required...)
 		slices.Sort(scopes)
 		scopes = slices.Compact(scopes)
@@ -410,6 +428,7 @@ func (s *Server) call(w http.ResponseWriter, r *http.Request, id json.RawMessage
 	case s.inflight <- struct{}{}:
 		defer func() { <-s.inflight }()
 	default:
+		mcpOutcomes["busy"].Inc()
 		writeResult(w, id, modern, toolError("FTW Cloud is busy. Try again in a moment."))
 		return
 	}
@@ -418,22 +437,29 @@ func (s *Server) call(w http.ResponseWriter, r *http.Request, id json.RawMessage
 		args = json.RawMessage("{}")
 	}
 	ctx := r.Context()
+	start := time.Now()
 	data, err := t.run(ctx, &callCtx{s: s, grant: g, rpcID: string(id)}, args)
+	mcpCall.Since(start)
 	if err != nil {
 		var ue userError
+		outcome := "tool_error"
 		if !errors.As(err, &ue) {
 			s.log.Warn("a tool call failed", "tool", t.name, "err", err)
 			ue = userError("The call failed inside FTW Cloud.")
+			outcome = "failed"
 		}
+		mcpOutcomes[outcome].Inc()
 		writeResult(w, id, modern, toolError(string(ue)))
 		return
 	}
 	if failure, ok := data.(failedToolData); ok {
+		mcpOutcomes["tool_error"].Inc()
 		result := toolResult(g.Subject.Home, failure.data)
 		result["isError"] = true
 		writeResult(w, id, modern, result)
 		return
 	}
+	mcpOutcomes["ok"].Inc()
 	writeResult(w, id, modern, toolResult(g.Subject.Home, data))
 }
 

internal/mcp/mcp_test.go

@@ -712,3 +712,18 @@ func TestARestartDrainsAgentSessions(t *testing.T) {
 		t.Fatalf("a drained instance opened a session: %d handshakes", hs)
 	}
 }
+
+// Calls are counted by outcome and timed, summed over every home and agent.
+func TestCallsAreCountedByOutcome(t *testing.T) {
+	w := newWorld(t)
+	b := w.connectBox()
+	a := w.connect(b, "")
+	ok, unknown, took := mcpOutcomes["ok"].Load(), mcpOutcomes["unknown_tool"].Load(), mcpCall.Count()
+	if _, isErr, text := a.call("status", nil); isErr {
+		t.Fatal(text)
+	}
+	_ = a.rpc("tools/call", map[string]any{"name": "no_such_tool"}, false)
+	if mcpOutcomes["ok"].Load() != ok+1 || mcpOutcomes["unknown_tool"].Load() != unknown+1 || mcpCall.Count() != took+1 {
+		t.Fatalf("ok %d, unknown %d, timed %d", mcpOutcomes["ok"].Load()-ok, mcpOutcomes["unknown_tool"].Load()-unknown, mcpCall.Count()-took)
+	}
+}